2
0

pthread_frame.c 32 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013
  1. /*
  2. * This file is part of FFmpeg.
  3. *
  4. * FFmpeg is free software; you can redistribute it and/or
  5. * modify it under the terms of the GNU Lesser General Public
  6. * License as published by the Free Software Foundation; either
  7. * version 2.1 of the License, or (at your option) any later version.
  8. *
  9. * FFmpeg is distributed in the hope that it will be useful,
  10. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  12. * Lesser General Public License for more details.
  13. *
  14. * You should have received a copy of the GNU Lesser General Public
  15. * License along with FFmpeg; if not, write to the Free Software
  16. * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  17. */
  18. /**
  19. * @file
  20. * Frame multithreading support functions
  21. * @see doc/multithreading.txt
  22. */
  23. #include "config.h"
  24. #include <stdatomic.h>
  25. #include <stdint.h>
  26. #include "avcodec.h"
  27. #include "hwaccel.h"
  28. #include "internal.h"
  29. #include "pthread_internal.h"
  30. #include "thread.h"
  31. #include "version.h"
  32. #include "libavutil/avassert.h"
  33. #include "libavutil/buffer.h"
  34. #include "libavutil/common.h"
  35. #include "libavutil/cpu.h"
  36. #include "libavutil/frame.h"
  37. #include "libavutil/internal.h"
  38. #include "libavutil/log.h"
  39. #include "libavutil/mem.h"
  40. #include "libavutil/opt.h"
  41. #include "libavutil/thread.h"
  42. enum {
  43. ///< Set when the thread is awaiting a packet.
  44. STATE_INPUT_READY,
  45. ///< Set before the codec has called ff_thread_finish_setup().
  46. STATE_SETTING_UP,
  47. /**
  48. * Set when the codec calls get_buffer().
  49. * State is returned to STATE_SETTING_UP afterwards.
  50. */
  51. STATE_GET_BUFFER,
  52. /**
  53. * Set when the codec calls get_format().
  54. * State is returned to STATE_SETTING_UP afterwards.
  55. */
  56. STATE_GET_FORMAT,
  57. ///< Set after the codec has called ff_thread_finish_setup().
  58. STATE_SETUP_FINISHED,
  59. };
  60. /**
  61. * Context used by codec threads and stored in their AVCodecInternal thread_ctx.
  62. */
  63. typedef struct PerThreadContext {
  64. struct FrameThreadContext *parent;
  65. pthread_t thread;
  66. int thread_init;
  67. pthread_cond_t input_cond; ///< Used to wait for a new packet from the main thread.
  68. pthread_cond_t progress_cond; ///< Used by child threads to wait for progress to change.
  69. pthread_cond_t output_cond; ///< Used by the main thread to wait for frames to finish.
  70. pthread_mutex_t mutex; ///< Mutex used to protect the contents of the PerThreadContext.
  71. pthread_mutex_t progress_mutex; ///< Mutex used to protect frame progress values and progress_cond.
  72. AVCodecContext *avctx; ///< Context used to decode packets passed to this thread.
  73. AVPacket avpkt; ///< Input packet (for decoding) or output (for encoding).
  74. AVFrame *frame; ///< Output frame (for decoding) or input (for encoding).
  75. int got_frame; ///< The output of got_picture_ptr from the last avcodec_decode_video() call.
  76. int result; ///< The result of the last codec decode/encode() call.
  77. atomic_int state;
  78. /**
  79. * Array of frames passed to ff_thread_release_buffer().
  80. * Frames are released after all threads referencing them are finished.
  81. */
  82. AVFrame *released_buffers;
  83. int num_released_buffers;
  84. int released_buffers_allocated;
  85. AVFrame *requested_frame; ///< AVFrame the codec passed to get_buffer()
  86. int requested_flags; ///< flags passed to get_buffer() for requested_frame
  87. const enum AVPixelFormat *available_formats; ///< Format array for get_format()
  88. enum AVPixelFormat result_format; ///< get_format() result
  89. int die; ///< Set when the thread should exit.
  90. int hwaccel_serializing;
  91. int async_serializing;
  92. atomic_int debug_threads; ///< Set if the FF_DEBUG_THREADS option is set.
  93. } PerThreadContext;
  94. /**
  95. * Context stored in the client AVCodecInternal thread_ctx.
  96. */
  97. typedef struct FrameThreadContext {
  98. PerThreadContext *threads; ///< The contexts for each thread.
  99. PerThreadContext *prev_thread; ///< The last thread submit_packet() was called on.
  100. pthread_mutex_t buffer_mutex; ///< Mutex used to protect get/release_buffer().
  101. /**
  102. * This lock is used for ensuring threads run in serial when hwaccel
  103. * is used.
  104. */
  105. pthread_mutex_t hwaccel_mutex;
  106. pthread_mutex_t async_mutex;
  107. pthread_cond_t async_cond;
  108. int async_lock;
  109. int next_decoding; ///< The next context to submit a packet to.
  110. int next_finished; ///< The next context to return output from.
  111. int delaying; /**<
  112. * Set for the first N packets, where N is the number of threads.
  113. * While it is set, ff_thread_en/decode_frame won't return any results.
  114. */
  115. } FrameThreadContext;
  116. #define THREAD_SAFE_CALLBACKS(avctx) \
  117. ((avctx)->thread_safe_callbacks || (avctx)->get_buffer2 == avcodec_default_get_buffer2)
  118. static void async_lock(FrameThreadContext *fctx)
  119. {
  120. pthread_mutex_lock(&fctx->async_mutex);
  121. while (fctx->async_lock)
  122. pthread_cond_wait(&fctx->async_cond, &fctx->async_mutex);
  123. fctx->async_lock = 1;
  124. pthread_mutex_unlock(&fctx->async_mutex);
  125. }
  126. static void async_unlock(FrameThreadContext *fctx)
  127. {
  128. pthread_mutex_lock(&fctx->async_mutex);
  129. av_assert0(fctx->async_lock);
  130. fctx->async_lock = 0;
  131. pthread_cond_broadcast(&fctx->async_cond);
  132. pthread_mutex_unlock(&fctx->async_mutex);
  133. }
  134. /**
  135. * Codec worker thread.
  136. *
  137. * Automatically calls ff_thread_finish_setup() if the codec does
  138. * not provide an update_thread_context method, or if the codec returns
  139. * before calling it.
  140. */
  141. static attribute_align_arg void *frame_worker_thread(void *arg)
  142. {
  143. PerThreadContext *p = arg;
  144. AVCodecContext *avctx = p->avctx;
  145. const AVCodec *codec = avctx->codec;
  146. pthread_mutex_lock(&p->mutex);
  147. while (1) {
  148. while (atomic_load(&p->state) == STATE_INPUT_READY && !p->die)
  149. pthread_cond_wait(&p->input_cond, &p->mutex);
  150. if (p->die) break;
  151. if (!codec->update_thread_context && THREAD_SAFE_CALLBACKS(avctx))
  152. ff_thread_finish_setup(avctx);
  153. /* If a decoder supports hwaccel, then it must call ff_get_format().
  154. * Since that call must happen before ff_thread_finish_setup(), the
  155. * decoder is required to implement update_thread_context() and call
  156. * ff_thread_finish_setup() manually. Therefore the above
  157. * ff_thread_finish_setup() call did not happen and hwaccel_serializing
  158. * cannot be true here. */
  159. av_assert0(!p->hwaccel_serializing);
  160. /* if the previous thread uses hwaccel then we take the lock to ensure
  161. * the threads don't run concurrently */
  162. if (avctx->hwaccel) {
  163. pthread_mutex_lock(&p->parent->hwaccel_mutex);
  164. p->hwaccel_serializing = 1;
  165. }
  166. av_frame_unref(p->frame);
  167. p->got_frame = 0;
  168. p->result = codec->decode(avctx, p->frame, &p->got_frame, &p->avpkt);
  169. if ((p->result < 0 || !p->got_frame) && p->frame->buf[0]) {
  170. if (avctx->internal->allocate_progress)
  171. av_log(avctx, AV_LOG_ERROR, "A frame threaded decoder did not "
  172. "free the frame on failure. This is a bug, please report it.\n");
  173. av_frame_unref(p->frame);
  174. }
  175. if (atomic_load(&p->state) == STATE_SETTING_UP)
  176. ff_thread_finish_setup(avctx);
  177. if (p->hwaccel_serializing) {
  178. p->hwaccel_serializing = 0;
  179. pthread_mutex_unlock(&p->parent->hwaccel_mutex);
  180. }
  181. if (p->async_serializing) {
  182. p->async_serializing = 0;
  183. async_unlock(p->parent);
  184. }
  185. pthread_mutex_lock(&p->progress_mutex);
  186. atomic_store(&p->state, STATE_INPUT_READY);
  187. pthread_cond_broadcast(&p->progress_cond);
  188. pthread_cond_signal(&p->output_cond);
  189. pthread_mutex_unlock(&p->progress_mutex);
  190. }
  191. pthread_mutex_unlock(&p->mutex);
  192. return NULL;
  193. }
  194. /**
  195. * Update the next thread's AVCodecContext with values from the reference thread's context.
  196. *
  197. * @param dst The destination context.
  198. * @param src The source context.
  199. * @param for_user 0 if the destination is a codec thread, 1 if the destination is the user's thread
  200. * @return 0 on success, negative error code on failure
  201. */
  202. static int update_context_from_thread(AVCodecContext *dst, AVCodecContext *src, int for_user)
  203. {
  204. int err = 0;
  205. if (dst != src && (for_user || !(src->codec_descriptor->props & AV_CODEC_PROP_INTRA_ONLY))) {
  206. dst->time_base = src->time_base;
  207. dst->framerate = src->framerate;
  208. dst->width = src->width;
  209. dst->height = src->height;
  210. dst->pix_fmt = src->pix_fmt;
  211. dst->sw_pix_fmt = src->sw_pix_fmt;
  212. dst->coded_width = src->coded_width;
  213. dst->coded_height = src->coded_height;
  214. dst->has_b_frames = src->has_b_frames;
  215. dst->idct_algo = src->idct_algo;
  216. dst->bits_per_coded_sample = src->bits_per_coded_sample;
  217. dst->sample_aspect_ratio = src->sample_aspect_ratio;
  218. dst->profile = src->profile;
  219. dst->level = src->level;
  220. dst->bits_per_raw_sample = src->bits_per_raw_sample;
  221. dst->ticks_per_frame = src->ticks_per_frame;
  222. dst->color_primaries = src->color_primaries;
  223. dst->color_trc = src->color_trc;
  224. dst->colorspace = src->colorspace;
  225. dst->color_range = src->color_range;
  226. dst->chroma_sample_location = src->chroma_sample_location;
  227. dst->hwaccel = src->hwaccel;
  228. dst->hwaccel_context = src->hwaccel_context;
  229. dst->channels = src->channels;
  230. dst->sample_rate = src->sample_rate;
  231. dst->sample_fmt = src->sample_fmt;
  232. dst->channel_layout = src->channel_layout;
  233. dst->internal->hwaccel_priv_data = src->internal->hwaccel_priv_data;
  234. if (!!dst->hw_frames_ctx != !!src->hw_frames_ctx ||
  235. (dst->hw_frames_ctx && dst->hw_frames_ctx->data != src->hw_frames_ctx->data)) {
  236. av_buffer_unref(&dst->hw_frames_ctx);
  237. if (src->hw_frames_ctx) {
  238. dst->hw_frames_ctx = av_buffer_ref(src->hw_frames_ctx);
  239. if (!dst->hw_frames_ctx)
  240. return AVERROR(ENOMEM);
  241. }
  242. }
  243. dst->hwaccel_flags = src->hwaccel_flags;
  244. }
  245. if (for_user) {
  246. dst->delay = src->thread_count - 1;
  247. #if FF_API_CODED_FRAME
  248. FF_DISABLE_DEPRECATION_WARNINGS
  249. dst->coded_frame = src->coded_frame;
  250. FF_ENABLE_DEPRECATION_WARNINGS
  251. #endif
  252. } else {
  253. if (dst->codec->update_thread_context)
  254. err = dst->codec->update_thread_context(dst, src);
  255. }
  256. return err;
  257. }
  258. /**
  259. * Update the next thread's AVCodecContext with values set by the user.
  260. *
  261. * @param dst The destination context.
  262. * @param src The source context.
  263. * @return 0 on success, negative error code on failure
  264. */
  265. static int update_context_from_user(AVCodecContext *dst, AVCodecContext *src)
  266. {
  267. #define copy_fields(s, e) memcpy(&dst->s, &src->s, (char*)&dst->e - (char*)&dst->s);
  268. dst->flags = src->flags;
  269. dst->draw_horiz_band= src->draw_horiz_band;
  270. dst->get_buffer2 = src->get_buffer2;
  271. dst->opaque = src->opaque;
  272. dst->debug = src->debug;
  273. dst->debug_mv = src->debug_mv;
  274. dst->slice_flags = src->slice_flags;
  275. dst->flags2 = src->flags2;
  276. copy_fields(skip_loop_filter, subtitle_header);
  277. dst->frame_number = src->frame_number;
  278. dst->reordered_opaque = src->reordered_opaque;
  279. dst->thread_safe_callbacks = src->thread_safe_callbacks;
  280. if (src->slice_count && src->slice_offset) {
  281. if (dst->slice_count < src->slice_count) {
  282. int err = av_reallocp_array(&dst->slice_offset, src->slice_count,
  283. sizeof(*dst->slice_offset));
  284. if (err < 0)
  285. return err;
  286. }
  287. memcpy(dst->slice_offset, src->slice_offset,
  288. src->slice_count * sizeof(*dst->slice_offset));
  289. }
  290. dst->slice_count = src->slice_count;
  291. return 0;
  292. #undef copy_fields
  293. }
  294. /// Releases the buffers that this decoding thread was the last user of.
  295. static void release_delayed_buffers(PerThreadContext *p)
  296. {
  297. FrameThreadContext *fctx = p->parent;
  298. while (p->num_released_buffers > 0) {
  299. AVFrame *f;
  300. pthread_mutex_lock(&fctx->buffer_mutex);
  301. // fix extended data in case the caller screwed it up
  302. av_assert0(p->avctx->codec_type == AVMEDIA_TYPE_VIDEO ||
  303. p->avctx->codec_type == AVMEDIA_TYPE_AUDIO);
  304. f = &p->released_buffers[--p->num_released_buffers];
  305. f->extended_data = f->data;
  306. av_frame_unref(f);
  307. pthread_mutex_unlock(&fctx->buffer_mutex);
  308. }
  309. }
  310. static int submit_packet(PerThreadContext *p, AVCodecContext *user_avctx,
  311. AVPacket *avpkt)
  312. {
  313. FrameThreadContext *fctx = p->parent;
  314. PerThreadContext *prev_thread = fctx->prev_thread;
  315. const AVCodec *codec = p->avctx->codec;
  316. int ret;
  317. if (!avpkt->size && !(codec->capabilities & AV_CODEC_CAP_DELAY))
  318. return 0;
  319. pthread_mutex_lock(&p->mutex);
  320. ret = update_context_from_user(p->avctx, user_avctx);
  321. if (ret) {
  322. pthread_mutex_unlock(&p->mutex);
  323. return ret;
  324. }
  325. atomic_store_explicit(&p->debug_threads,
  326. (p->avctx->debug & FF_DEBUG_THREADS) != 0,
  327. memory_order_relaxed);
  328. release_delayed_buffers(p);
  329. if (prev_thread) {
  330. int err;
  331. if (atomic_load(&prev_thread->state) == STATE_SETTING_UP) {
  332. pthread_mutex_lock(&prev_thread->progress_mutex);
  333. while (atomic_load(&prev_thread->state) == STATE_SETTING_UP)
  334. pthread_cond_wait(&prev_thread->progress_cond, &prev_thread->progress_mutex);
  335. pthread_mutex_unlock(&prev_thread->progress_mutex);
  336. }
  337. err = update_context_from_thread(p->avctx, prev_thread->avctx, 0);
  338. if (err) {
  339. pthread_mutex_unlock(&p->mutex);
  340. return err;
  341. }
  342. }
  343. av_packet_unref(&p->avpkt);
  344. ret = av_packet_ref(&p->avpkt, avpkt);
  345. if (ret < 0) {
  346. pthread_mutex_unlock(&p->mutex);
  347. av_log(p->avctx, AV_LOG_ERROR, "av_packet_ref() failed in submit_packet()\n");
  348. return ret;
  349. }
  350. atomic_store(&p->state, STATE_SETTING_UP);
  351. pthread_cond_signal(&p->input_cond);
  352. pthread_mutex_unlock(&p->mutex);
  353. /*
  354. * If the client doesn't have a thread-safe get_buffer(),
  355. * then decoding threads call back to the main thread,
  356. * and it calls back to the client here.
  357. */
  358. if (!p->avctx->thread_safe_callbacks && (
  359. p->avctx->get_format != avcodec_default_get_format ||
  360. p->avctx->get_buffer2 != avcodec_default_get_buffer2)) {
  361. while (atomic_load(&p->state) != STATE_SETUP_FINISHED && atomic_load(&p->state) != STATE_INPUT_READY) {
  362. int call_done = 1;
  363. pthread_mutex_lock(&p->progress_mutex);
  364. while (atomic_load(&p->state) == STATE_SETTING_UP)
  365. pthread_cond_wait(&p->progress_cond, &p->progress_mutex);
  366. switch (atomic_load_explicit(&p->state, memory_order_acquire)) {
  367. case STATE_GET_BUFFER:
  368. p->result = ff_get_buffer(p->avctx, p->requested_frame, p->requested_flags);
  369. break;
  370. case STATE_GET_FORMAT:
  371. p->result_format = ff_get_format(p->avctx, p->available_formats);
  372. break;
  373. default:
  374. call_done = 0;
  375. break;
  376. }
  377. if (call_done) {
  378. atomic_store(&p->state, STATE_SETTING_UP);
  379. pthread_cond_signal(&p->progress_cond);
  380. }
  381. pthread_mutex_unlock(&p->progress_mutex);
  382. }
  383. }
  384. fctx->prev_thread = p;
  385. fctx->next_decoding++;
  386. return 0;
  387. }
  388. int ff_thread_decode_frame(AVCodecContext *avctx,
  389. AVFrame *picture, int *got_picture_ptr,
  390. AVPacket *avpkt)
  391. {
  392. FrameThreadContext *fctx = avctx->internal->thread_ctx;
  393. int finished = fctx->next_finished;
  394. PerThreadContext *p;
  395. int err;
  396. /* release the async lock, permitting blocked hwaccel threads to
  397. * go forward while we are in this function */
  398. async_unlock(fctx);
  399. /*
  400. * Submit a packet to the next decoding thread.
  401. */
  402. p = &fctx->threads[fctx->next_decoding];
  403. err = submit_packet(p, avctx, avpkt);
  404. if (err)
  405. goto finish;
  406. /*
  407. * If we're still receiving the initial packets, don't return a frame.
  408. */
  409. if (fctx->next_decoding > (avctx->thread_count-1-(avctx->codec_id == AV_CODEC_ID_FFV1)))
  410. fctx->delaying = 0;
  411. if (fctx->delaying) {
  412. *got_picture_ptr=0;
  413. if (avpkt->size) {
  414. err = avpkt->size;
  415. goto finish;
  416. }
  417. }
  418. /*
  419. * Return the next available frame from the oldest thread.
  420. * If we're at the end of the stream, then we have to skip threads that
  421. * didn't output a frame/error, because we don't want to accidentally signal
  422. * EOF (avpkt->size == 0 && *got_picture_ptr == 0 && err >= 0).
  423. */
  424. do {
  425. p = &fctx->threads[finished++];
  426. if (atomic_load(&p->state) != STATE_INPUT_READY) {
  427. pthread_mutex_lock(&p->progress_mutex);
  428. while (atomic_load_explicit(&p->state, memory_order_relaxed) != STATE_INPUT_READY)
  429. pthread_cond_wait(&p->output_cond, &p->progress_mutex);
  430. pthread_mutex_unlock(&p->progress_mutex);
  431. }
  432. av_frame_move_ref(picture, p->frame);
  433. *got_picture_ptr = p->got_frame;
  434. picture->pkt_dts = p->avpkt.dts;
  435. err = p->result;
  436. /*
  437. * A later call with avkpt->size == 0 may loop over all threads,
  438. * including this one, searching for a frame/error to return before being
  439. * stopped by the "finished != fctx->next_finished" condition.
  440. * Make sure we don't mistakenly return the same frame/error again.
  441. */
  442. p->got_frame = 0;
  443. p->result = 0;
  444. if (finished >= avctx->thread_count) finished = 0;
  445. } while (!avpkt->size && !*got_picture_ptr && err >= 0 && finished != fctx->next_finished);
  446. update_context_from_thread(avctx, p->avctx, 1);
  447. if (fctx->next_decoding >= avctx->thread_count) fctx->next_decoding = 0;
  448. fctx->next_finished = finished;
  449. /* return the size of the consumed packet if no error occurred */
  450. if (err >= 0)
  451. err = avpkt->size;
  452. finish:
  453. async_lock(fctx);
  454. return err;
  455. }
  456. void ff_thread_report_progress(ThreadFrame *f, int n, int field)
  457. {
  458. PerThreadContext *p;
  459. atomic_int *progress = f->progress ? (atomic_int*)f->progress->data : NULL;
  460. if (!progress ||
  461. atomic_load_explicit(&progress[field], memory_order_relaxed) >= n)
  462. return;
  463. p = f->owner[field]->internal->thread_ctx;
  464. if (atomic_load_explicit(&p->debug_threads, memory_order_relaxed))
  465. av_log(f->owner[field], AV_LOG_DEBUG,
  466. "%p finished %d field %d\n", progress, n, field);
  467. pthread_mutex_lock(&p->progress_mutex);
  468. atomic_store_explicit(&progress[field], n, memory_order_release);
  469. pthread_cond_broadcast(&p->progress_cond);
  470. pthread_mutex_unlock(&p->progress_mutex);
  471. }
  472. void ff_thread_await_progress(ThreadFrame *f, int n, int field)
  473. {
  474. PerThreadContext *p;
  475. atomic_int *progress = f->progress ? (atomic_int*)f->progress->data : NULL;
  476. if (!progress ||
  477. atomic_load_explicit(&progress[field], memory_order_acquire) >= n)
  478. return;
  479. p = f->owner[field]->internal->thread_ctx;
  480. if (atomic_load_explicit(&p->debug_threads, memory_order_relaxed))
  481. av_log(f->owner[field], AV_LOG_DEBUG,
  482. "thread awaiting %d field %d from %p\n", n, field, progress);
  483. pthread_mutex_lock(&p->progress_mutex);
  484. while (atomic_load_explicit(&progress[field], memory_order_relaxed) < n)
  485. pthread_cond_wait(&p->progress_cond, &p->progress_mutex);
  486. pthread_mutex_unlock(&p->progress_mutex);
  487. }
  488. void ff_thread_finish_setup(AVCodecContext *avctx) {
  489. PerThreadContext *p = avctx->internal->thread_ctx;
  490. if (!(avctx->active_thread_type&FF_THREAD_FRAME)) return;
  491. if (avctx->hwaccel && !p->hwaccel_serializing) {
  492. pthread_mutex_lock(&p->parent->hwaccel_mutex);
  493. p->hwaccel_serializing = 1;
  494. }
  495. /* this assumes that no hwaccel calls happen before ff_thread_finish_setup() */
  496. if (avctx->hwaccel &&
  497. !(avctx->hwaccel->caps_internal & HWACCEL_CAP_ASYNC_SAFE)) {
  498. p->async_serializing = 1;
  499. async_lock(p->parent);
  500. }
  501. pthread_mutex_lock(&p->progress_mutex);
  502. if(atomic_load(&p->state) == STATE_SETUP_FINISHED){
  503. av_log(avctx, AV_LOG_WARNING, "Multiple ff_thread_finish_setup() calls\n");
  504. }
  505. atomic_store(&p->state, STATE_SETUP_FINISHED);
  506. pthread_cond_broadcast(&p->progress_cond);
  507. pthread_mutex_unlock(&p->progress_mutex);
  508. }
  509. /// Waits for all threads to finish.
  510. static void park_frame_worker_threads(FrameThreadContext *fctx, int thread_count)
  511. {
  512. int i;
  513. async_unlock(fctx);
  514. for (i = 0; i < thread_count; i++) {
  515. PerThreadContext *p = &fctx->threads[i];
  516. if (atomic_load(&p->state) != STATE_INPUT_READY) {
  517. pthread_mutex_lock(&p->progress_mutex);
  518. while (atomic_load(&p->state) != STATE_INPUT_READY)
  519. pthread_cond_wait(&p->output_cond, &p->progress_mutex);
  520. pthread_mutex_unlock(&p->progress_mutex);
  521. }
  522. p->got_frame = 0;
  523. }
  524. async_lock(fctx);
  525. }
  526. void ff_frame_thread_free(AVCodecContext *avctx, int thread_count)
  527. {
  528. FrameThreadContext *fctx = avctx->internal->thread_ctx;
  529. const AVCodec *codec = avctx->codec;
  530. int i;
  531. park_frame_worker_threads(fctx, thread_count);
  532. if (fctx->prev_thread && fctx->prev_thread != fctx->threads)
  533. if (update_context_from_thread(fctx->threads->avctx, fctx->prev_thread->avctx, 0) < 0) {
  534. av_log(avctx, AV_LOG_ERROR, "Final thread update failed\n");
  535. fctx->prev_thread->avctx->internal->is_copy = fctx->threads->avctx->internal->is_copy;
  536. fctx->threads->avctx->internal->is_copy = 1;
  537. }
  538. for (i = 0; i < thread_count; i++) {
  539. PerThreadContext *p = &fctx->threads[i];
  540. pthread_mutex_lock(&p->mutex);
  541. p->die = 1;
  542. pthread_cond_signal(&p->input_cond);
  543. pthread_mutex_unlock(&p->mutex);
  544. if (p->thread_init)
  545. pthread_join(p->thread, NULL);
  546. p->thread_init=0;
  547. if (codec->close && p->avctx)
  548. codec->close(p->avctx);
  549. release_delayed_buffers(p);
  550. av_frame_free(&p->frame);
  551. }
  552. for (i = 0; i < thread_count; i++) {
  553. PerThreadContext *p = &fctx->threads[i];
  554. pthread_mutex_destroy(&p->mutex);
  555. pthread_mutex_destroy(&p->progress_mutex);
  556. pthread_cond_destroy(&p->input_cond);
  557. pthread_cond_destroy(&p->progress_cond);
  558. pthread_cond_destroy(&p->output_cond);
  559. av_packet_unref(&p->avpkt);
  560. av_freep(&p->released_buffers);
  561. if (i && p->avctx) {
  562. av_freep(&p->avctx->priv_data);
  563. av_freep(&p->avctx->slice_offset);
  564. }
  565. if (p->avctx) {
  566. av_freep(&p->avctx->internal);
  567. av_buffer_unref(&p->avctx->hw_frames_ctx);
  568. }
  569. av_freep(&p->avctx);
  570. }
  571. av_freep(&fctx->threads);
  572. pthread_mutex_destroy(&fctx->buffer_mutex);
  573. pthread_mutex_destroy(&fctx->hwaccel_mutex);
  574. pthread_mutex_destroy(&fctx->async_mutex);
  575. pthread_cond_destroy(&fctx->async_cond);
  576. av_freep(&avctx->internal->thread_ctx);
  577. if (avctx->priv_data && avctx->codec && avctx->codec->priv_class)
  578. av_opt_free(avctx->priv_data);
  579. avctx->codec = NULL;
  580. }
  581. int ff_frame_thread_init(AVCodecContext *avctx)
  582. {
  583. int thread_count = avctx->thread_count;
  584. const AVCodec *codec = avctx->codec;
  585. AVCodecContext *src = avctx;
  586. FrameThreadContext *fctx;
  587. int i, err = 0;
  588. if (!thread_count) {
  589. int nb_cpus = av_cpu_count();
  590. #if FF_API_DEBUG_MV
  591. if ((avctx->debug & (FF_DEBUG_VIS_QP | FF_DEBUG_VIS_MB_TYPE)) || avctx->debug_mv)
  592. nb_cpus = 1;
  593. #endif
  594. // use number of cores + 1 as thread count if there is more than one
  595. if (nb_cpus > 1)
  596. thread_count = avctx->thread_count = FFMIN(nb_cpus + 1, MAX_AUTO_THREADS);
  597. else
  598. thread_count = avctx->thread_count = 1;
  599. }
  600. if (thread_count <= 1) {
  601. avctx->active_thread_type = 0;
  602. return 0;
  603. }
  604. avctx->internal->thread_ctx = fctx = av_mallocz(sizeof(FrameThreadContext));
  605. if (!fctx)
  606. return AVERROR(ENOMEM);
  607. fctx->threads = av_mallocz_array(thread_count, sizeof(PerThreadContext));
  608. if (!fctx->threads) {
  609. av_freep(&avctx->internal->thread_ctx);
  610. return AVERROR(ENOMEM);
  611. }
  612. pthread_mutex_init(&fctx->buffer_mutex, NULL);
  613. pthread_mutex_init(&fctx->hwaccel_mutex, NULL);
  614. pthread_mutex_init(&fctx->async_mutex, NULL);
  615. pthread_cond_init(&fctx->async_cond, NULL);
  616. fctx->async_lock = 1;
  617. fctx->delaying = 1;
  618. for (i = 0; i < thread_count; i++) {
  619. AVCodecContext *copy = av_malloc(sizeof(AVCodecContext));
  620. PerThreadContext *p = &fctx->threads[i];
  621. pthread_mutex_init(&p->mutex, NULL);
  622. pthread_mutex_init(&p->progress_mutex, NULL);
  623. pthread_cond_init(&p->input_cond, NULL);
  624. pthread_cond_init(&p->progress_cond, NULL);
  625. pthread_cond_init(&p->output_cond, NULL);
  626. p->frame = av_frame_alloc();
  627. if (!p->frame) {
  628. av_freep(&copy);
  629. err = AVERROR(ENOMEM);
  630. goto error;
  631. }
  632. p->parent = fctx;
  633. p->avctx = copy;
  634. if (!copy) {
  635. err = AVERROR(ENOMEM);
  636. goto error;
  637. }
  638. *copy = *src;
  639. copy->internal = av_malloc(sizeof(AVCodecInternal));
  640. if (!copy->internal) {
  641. copy->priv_data = NULL;
  642. err = AVERROR(ENOMEM);
  643. goto error;
  644. }
  645. *copy->internal = *src->internal;
  646. copy->internal->thread_ctx = p;
  647. copy->internal->last_pkt_props = &p->avpkt;
  648. if (!i) {
  649. src = copy;
  650. if (codec->init)
  651. err = codec->init(copy);
  652. update_context_from_thread(avctx, copy, 1);
  653. } else {
  654. copy->priv_data = av_malloc(codec->priv_data_size);
  655. if (!copy->priv_data) {
  656. err = AVERROR(ENOMEM);
  657. goto error;
  658. }
  659. memcpy(copy->priv_data, src->priv_data, codec->priv_data_size);
  660. copy->internal->is_copy = 1;
  661. if (codec->init_thread_copy)
  662. err = codec->init_thread_copy(copy);
  663. }
  664. if (err) goto error;
  665. atomic_init(&p->debug_threads, (copy->debug & FF_DEBUG_THREADS) != 0);
  666. err = AVERROR(pthread_create(&p->thread, NULL, frame_worker_thread, p));
  667. p->thread_init= !err;
  668. if(!p->thread_init)
  669. goto error;
  670. }
  671. return 0;
  672. error:
  673. ff_frame_thread_free(avctx, i+1);
  674. return err;
  675. }
  676. void ff_thread_flush(AVCodecContext *avctx)
  677. {
  678. int i;
  679. FrameThreadContext *fctx = avctx->internal->thread_ctx;
  680. if (!fctx) return;
  681. park_frame_worker_threads(fctx, avctx->thread_count);
  682. if (fctx->prev_thread) {
  683. if (fctx->prev_thread != &fctx->threads[0])
  684. update_context_from_thread(fctx->threads[0].avctx, fctx->prev_thread->avctx, 0);
  685. }
  686. fctx->next_decoding = fctx->next_finished = 0;
  687. fctx->delaying = 1;
  688. fctx->prev_thread = NULL;
  689. for (i = 0; i < avctx->thread_count; i++) {
  690. PerThreadContext *p = &fctx->threads[i];
  691. // Make sure decode flush calls with size=0 won't return old frames
  692. p->got_frame = 0;
  693. av_frame_unref(p->frame);
  694. p->result = 0;
  695. release_delayed_buffers(p);
  696. if (avctx->codec->flush)
  697. avctx->codec->flush(p->avctx);
  698. }
  699. }
  700. int ff_thread_can_start_frame(AVCodecContext *avctx)
  701. {
  702. PerThreadContext *p = avctx->internal->thread_ctx;
  703. if ((avctx->active_thread_type&FF_THREAD_FRAME) && atomic_load(&p->state) != STATE_SETTING_UP &&
  704. (avctx->codec->update_thread_context || !THREAD_SAFE_CALLBACKS(avctx))) {
  705. return 0;
  706. }
  707. return 1;
  708. }
  709. static int thread_get_buffer_internal(AVCodecContext *avctx, ThreadFrame *f, int flags)
  710. {
  711. PerThreadContext *p = avctx->internal->thread_ctx;
  712. int err;
  713. f->owner[0] = f->owner[1] = avctx;
  714. if (!(avctx->active_thread_type & FF_THREAD_FRAME))
  715. return ff_get_buffer(avctx, f->f, flags);
  716. if (atomic_load(&p->state) != STATE_SETTING_UP &&
  717. (avctx->codec->update_thread_context || !THREAD_SAFE_CALLBACKS(avctx))) {
  718. av_log(avctx, AV_LOG_ERROR, "get_buffer() cannot be called after ff_thread_finish_setup()\n");
  719. return -1;
  720. }
  721. if (avctx->internal->allocate_progress) {
  722. atomic_int *progress;
  723. f->progress = av_buffer_alloc(2 * sizeof(*progress));
  724. if (!f->progress) {
  725. return AVERROR(ENOMEM);
  726. }
  727. progress = (atomic_int*)f->progress->data;
  728. atomic_init(&progress[0], -1);
  729. atomic_init(&progress[1], -1);
  730. }
  731. pthread_mutex_lock(&p->parent->buffer_mutex);
  732. if (THREAD_SAFE_CALLBACKS(avctx)) {
  733. err = ff_get_buffer(avctx, f->f, flags);
  734. } else {
  735. pthread_mutex_lock(&p->progress_mutex);
  736. p->requested_frame = f->f;
  737. p->requested_flags = flags;
  738. atomic_store_explicit(&p->state, STATE_GET_BUFFER, memory_order_release);
  739. pthread_cond_broadcast(&p->progress_cond);
  740. while (atomic_load(&p->state) != STATE_SETTING_UP)
  741. pthread_cond_wait(&p->progress_cond, &p->progress_mutex);
  742. err = p->result;
  743. pthread_mutex_unlock(&p->progress_mutex);
  744. }
  745. if (!THREAD_SAFE_CALLBACKS(avctx) && !avctx->codec->update_thread_context)
  746. ff_thread_finish_setup(avctx);
  747. if (err)
  748. av_buffer_unref(&f->progress);
  749. pthread_mutex_unlock(&p->parent->buffer_mutex);
  750. return err;
  751. }
  752. enum AVPixelFormat ff_thread_get_format(AVCodecContext *avctx, const enum AVPixelFormat *fmt)
  753. {
  754. enum AVPixelFormat res;
  755. PerThreadContext *p = avctx->internal->thread_ctx;
  756. if (!(avctx->active_thread_type & FF_THREAD_FRAME) || avctx->thread_safe_callbacks ||
  757. avctx->get_format == avcodec_default_get_format)
  758. return ff_get_format(avctx, fmt);
  759. if (atomic_load(&p->state) != STATE_SETTING_UP) {
  760. av_log(avctx, AV_LOG_ERROR, "get_format() cannot be called after ff_thread_finish_setup()\n");
  761. return -1;
  762. }
  763. pthread_mutex_lock(&p->progress_mutex);
  764. p->available_formats = fmt;
  765. atomic_store(&p->state, STATE_GET_FORMAT);
  766. pthread_cond_broadcast(&p->progress_cond);
  767. while (atomic_load(&p->state) != STATE_SETTING_UP)
  768. pthread_cond_wait(&p->progress_cond, &p->progress_mutex);
  769. res = p->result_format;
  770. pthread_mutex_unlock(&p->progress_mutex);
  771. return res;
  772. }
  773. int ff_thread_get_buffer(AVCodecContext *avctx, ThreadFrame *f, int flags)
  774. {
  775. int ret = thread_get_buffer_internal(avctx, f, flags);
  776. if (ret < 0)
  777. av_log(avctx, AV_LOG_ERROR, "thread_get_buffer() failed\n");
  778. return ret;
  779. }
  780. void ff_thread_release_buffer(AVCodecContext *avctx, ThreadFrame *f)
  781. {
  782. PerThreadContext *p = avctx->internal->thread_ctx;
  783. FrameThreadContext *fctx;
  784. AVFrame *dst, *tmp;
  785. int can_direct_free = !(avctx->active_thread_type & FF_THREAD_FRAME) ||
  786. THREAD_SAFE_CALLBACKS(avctx);
  787. if (!f->f || !f->f->buf[0])
  788. return;
  789. if (avctx->debug & FF_DEBUG_BUFFERS)
  790. av_log(avctx, AV_LOG_DEBUG, "thread_release_buffer called on pic %p\n", f);
  791. av_buffer_unref(&f->progress);
  792. f->owner[0] = f->owner[1] = NULL;
  793. if (can_direct_free) {
  794. av_frame_unref(f->f);
  795. return;
  796. }
  797. fctx = p->parent;
  798. pthread_mutex_lock(&fctx->buffer_mutex);
  799. if (p->num_released_buffers + 1 >= INT_MAX / sizeof(*p->released_buffers))
  800. goto fail;
  801. tmp = av_fast_realloc(p->released_buffers, &p->released_buffers_allocated,
  802. (p->num_released_buffers + 1) *
  803. sizeof(*p->released_buffers));
  804. if (!tmp)
  805. goto fail;
  806. p->released_buffers = tmp;
  807. dst = &p->released_buffers[p->num_released_buffers];
  808. av_frame_move_ref(dst, f->f);
  809. p->num_released_buffers++;
  810. fail:
  811. pthread_mutex_unlock(&fctx->buffer_mutex);
  812. }