api-threadmessage-test.c 9.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260
  1. /*
  2. * Permission is hereby granted, free of charge, to any person obtaining a copy
  3. * of this software and associated documentation files (the "Software"), to deal
  4. * in the Software without restriction, including without limitation the rights
  5. * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  6. * copies of the Software, and to permit persons to whom the Software is
  7. * furnished to do so, subject to the following conditions:
  8. *
  9. * The above copyright notice and this permission notice shall be included in
  10. * all copies or substantial portions of the Software.
  11. *
  12. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  13. * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  14. * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
  15. * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  16. * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  17. * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  18. * THE SOFTWARE.
  19. */
  20. /**
  21. * Thread message API test
  22. */
  23. #include "libavutil/avassert.h"
  24. #include "libavutil/avstring.h"
  25. #include "libavutil/frame.h"
  26. #include "libavutil/threadmessage.h"
  27. #include "libavutil/thread.h" // not public
  28. struct sender_data {
  29. int id;
  30. pthread_t tid;
  31. int workload;
  32. AVThreadMessageQueue *queue;
  33. };
  34. /* same as sender_data but shuffled for testing purpose */
  35. struct receiver_data {
  36. pthread_t tid;
  37. int workload;
  38. int id;
  39. AVThreadMessageQueue *queue;
  40. };
  41. struct message {
  42. AVFrame *frame;
  43. // we add some junk in the message to make sure the message size is >
  44. // sizeof(void*)
  45. int magic;
  46. };
  47. #define MAGIC 0xdeadc0de
  48. static void free_frame(void *arg)
  49. {
  50. struct message *msg = arg;
  51. av_assert0(msg->magic == MAGIC);
  52. av_frame_free(&msg->frame);
  53. }
  54. static void *sender_thread(void *arg)
  55. {
  56. int i, ret = 0;
  57. struct sender_data *wd = arg;
  58. av_log(NULL, AV_LOG_INFO, "sender #%d: workload=%d\n", wd->id, wd->workload);
  59. for (i = 0; i < wd->workload; i++) {
  60. if (rand() % wd->workload < wd->workload / 10) {
  61. av_log(NULL, AV_LOG_INFO, "sender #%d: flushing the queue\n", wd->id);
  62. av_thread_message_flush(wd->queue);
  63. } else {
  64. char *val;
  65. AVDictionary *meta = NULL;
  66. struct message msg = {
  67. .magic = MAGIC,
  68. .frame = av_frame_alloc(),
  69. };
  70. if (!msg.frame) {
  71. ret = AVERROR(ENOMEM);
  72. break;
  73. }
  74. /* we add some metadata to identify the frames */
  75. val = av_asprintf("frame %d/%d from sender %d",
  76. i + 1, wd->workload, wd->id);
  77. if (!val) {
  78. av_frame_free(&msg.frame);
  79. ret = AVERROR(ENOMEM);
  80. break;
  81. }
  82. ret = av_dict_set(&meta, "sig", val, AV_DICT_DONT_STRDUP_VAL);
  83. if (ret < 0) {
  84. av_frame_free(&msg.frame);
  85. break;
  86. }
  87. av_frame_set_metadata(msg.frame, meta);
  88. /* allocate a real frame in order to simulate "real" work */
  89. msg.frame->format = AV_PIX_FMT_RGBA;
  90. msg.frame->width = 320;
  91. msg.frame->height = 240;
  92. ret = av_frame_get_buffer(msg.frame, 32);
  93. if (ret < 0) {
  94. av_frame_free(&msg.frame);
  95. break;
  96. }
  97. /* push the frame in the common queue */
  98. av_log(NULL, AV_LOG_INFO, "sender #%d: sending my work (%d/%d frame:%p)\n",
  99. wd->id, i + 1, wd->workload, msg.frame);
  100. ret = av_thread_message_queue_send(wd->queue, &msg, 0);
  101. if (ret < 0) {
  102. av_frame_free(&msg.frame);
  103. break;
  104. }
  105. }
  106. }
  107. av_log(NULL, AV_LOG_INFO, "sender #%d: my work is done here (%s)\n",
  108. wd->id, av_err2str(ret));
  109. av_thread_message_queue_set_err_recv(wd->queue, ret < 0 ? ret : AVERROR_EOF);
  110. return NULL;
  111. }
  112. static void *receiver_thread(void *arg)
  113. {
  114. int i, ret = 0;
  115. struct receiver_data *rd = arg;
  116. for (i = 0; i < rd->workload; i++) {
  117. if (rand() % rd->workload < rd->workload / 10) {
  118. av_log(NULL, AV_LOG_INFO, "receiver #%d: flushing the queue\n", rd->id);
  119. av_thread_message_flush(rd->queue);
  120. } else {
  121. struct message msg;
  122. AVDictionary *meta;
  123. AVDictionaryEntry *e;
  124. ret = av_thread_message_queue_recv(rd->queue, &msg, 0);
  125. if (ret < 0)
  126. break;
  127. av_assert0(msg.magic == MAGIC);
  128. meta = av_frame_get_metadata(msg.frame);
  129. e = av_dict_get(meta, "sig", NULL, 0);
  130. av_log(NULL, AV_LOG_INFO, "got \"%s\" (%p)\n", e->value, msg.frame);
  131. av_frame_free(&msg.frame);
  132. }
  133. }
  134. av_log(NULL, AV_LOG_INFO, "consumed enough (%d), stop\n", i);
  135. av_thread_message_queue_set_err_send(rd->queue, ret < 0 ? ret : AVERROR_EOF);
  136. return NULL;
  137. }
  138. static int get_workload(int minv, int maxv)
  139. {
  140. return maxv == minv ? maxv : rand() % (maxv - minv) + minv;
  141. }
  142. int main(int ac, char **av)
  143. {
  144. int i, ret = 0;
  145. int max_queue_size;
  146. int nb_senders, sender_min_load, sender_max_load;
  147. int nb_receivers, receiver_min_load, receiver_max_load;
  148. struct sender_data *senders;
  149. struct receiver_data *receivers;
  150. AVThreadMessageQueue *queue = NULL;
  151. if (ac != 8) {
  152. av_log(NULL, AV_LOG_ERROR, "%s <max_queue_size> "
  153. "<nb_senders> <sender_min_send> <sender_max_send> "
  154. "<nb_receivers> <receiver_min_recv> <receiver_max_recv>\n", av[0]);
  155. return 1;
  156. }
  157. max_queue_size = atoi(av[1]);
  158. nb_senders = atoi(av[2]);
  159. sender_min_load = atoi(av[3]);
  160. sender_max_load = atoi(av[4]);
  161. nb_receivers = atoi(av[5]);
  162. receiver_min_load = atoi(av[6]);
  163. receiver_max_load = atoi(av[7]);
  164. if (max_queue_size <= 0 ||
  165. nb_senders <= 0 || sender_min_load <= 0 || sender_max_load <= 0 ||
  166. nb_receivers <= 0 || receiver_min_load <= 0 || receiver_max_load <= 0) {
  167. av_log(NULL, AV_LOG_ERROR, "negative values not allowed\n");
  168. return 1;
  169. }
  170. av_log(NULL, AV_LOG_INFO, "qsize:%d / %d senders sending [%d-%d] / "
  171. "%d receivers receiving [%d-%d]\n", max_queue_size,
  172. nb_senders, sender_min_load, sender_max_load,
  173. nb_receivers, receiver_min_load, receiver_max_load);
  174. senders = av_mallocz_array(nb_senders, sizeof(*senders));
  175. receivers = av_mallocz_array(nb_receivers, sizeof(*receivers));
  176. if (!senders || !receivers) {
  177. ret = AVERROR(ENOMEM);
  178. goto end;
  179. }
  180. ret = av_thread_message_queue_alloc(&queue, max_queue_size, sizeof(struct message));
  181. if (ret < 0)
  182. goto end;
  183. av_thread_message_queue_set_free_func(queue, free_frame);
  184. #define SPAWN_THREADS(type) do { \
  185. for (i = 0; i < nb_##type##s; i++) { \
  186. struct type##_data *td = &type##s[i]; \
  187. \
  188. td->id = i; \
  189. td->queue = queue; \
  190. td->workload = get_workload(type##_min_load, type##_max_load); \
  191. \
  192. ret = pthread_create(&td->tid, NULL, type##_thread, td); \
  193. if (ret) { \
  194. const int err = AVERROR(ret); \
  195. av_log(NULL, AV_LOG_ERROR, "Unable to start " AV_STRINGIFY(type) \
  196. " thread: %s\n", av_err2str(err)); \
  197. goto end; \
  198. } \
  199. } \
  200. } while (0)
  201. #define WAIT_THREADS(type) do { \
  202. for (i = 0; i < nb_##type##s; i++) { \
  203. struct type##_data *td = &type##s[i]; \
  204. \
  205. ret = pthread_join(td->tid, NULL); \
  206. if (ret) { \
  207. const int err = AVERROR(ret); \
  208. av_log(NULL, AV_LOG_ERROR, "Unable to join " AV_STRINGIFY(type) \
  209. " thread: %s\n", av_err2str(err)); \
  210. goto end; \
  211. } \
  212. } \
  213. } while (0)
  214. SPAWN_THREADS(receiver);
  215. SPAWN_THREADS(sender);
  216. WAIT_THREADS(sender);
  217. WAIT_THREADS(receiver);
  218. end:
  219. av_thread_message_queue_free(&queue);
  220. av_freep(&senders);
  221. av_freep(&receivers);
  222. if (ret < 0 && ret != AVERROR_EOF) {
  223. av_log(NULL, AV_LOG_ERROR, "Error: %s\n", av_err2str(ret));
  224. return 1;
  225. }
  226. return 0;
  227. }