rtmp-stream.c 42 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640
  1. /******************************************************************************
  2. Copyright (C) 2014 by Hugh Bailey <[email protected]>
  3. This program is free software: you can redistribute it and/or modify
  4. it under the terms of the GNU General Public License as published by
  5. the Free Software Foundation, either version 2 of the License, or
  6. (at your option) any later version.
  7. This program is distributed in the hope that it will be useful,
  8. but WITHOUT ANY WARRANTY; without even the implied warranty of
  9. MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  10. GNU General Public License for more details.
  11. You should have received a copy of the GNU General Public License
  12. along with this program. If not, see <http://www.gnu.org/licenses/>.
  13. ******************************************************************************/
  14. #include "rtmp-stream.h"
  15. #ifdef _WIN32
  16. #include <util/windows/win-version.h>
  17. #endif
  18. #ifndef SEC_TO_NSEC
  19. #define SEC_TO_NSEC 1000000000ULL
  20. #endif
  21. #ifndef MSEC_TO_USEC
  22. #define MSEC_TO_USEC 1000ULL
  23. #endif
  24. #ifndef MSEC_TO_NSEC
  25. #define MSEC_TO_NSEC 1000000ULL
  26. #endif
  27. /* dynamic bitrate coefficients */
  28. #define DBR_INC_TIMER (30ULL * SEC_TO_NSEC)
  29. #define DBR_TRIGGER_USEC (200ULL * MSEC_TO_USEC)
  30. #define MIN_ESTIMATE_DURATION_MS 1000
  31. #define MAX_ESTIMATE_DURATION_MS 2000
  32. static const char *rtmp_stream_getname(void *unused)
  33. {
  34. UNUSED_PARAMETER(unused);
  35. return obs_module_text("RTMPStream");
  36. }
  37. static void log_rtmp(int level, const char *format, va_list args)
  38. {
  39. if (level > RTMP_LOGWARNING)
  40. return;
  41. blogva(LOG_INFO, format, args);
  42. }
  43. static inline size_t num_buffered_packets(struct rtmp_stream *stream);
  44. static inline void free_packets(struct rtmp_stream *stream)
  45. {
  46. size_t num_packets;
  47. pthread_mutex_lock(&stream->packets_mutex);
  48. num_packets = num_buffered_packets(stream);
  49. if (num_packets)
  50. info("Freeing %d remaining packets", (int)num_packets);
  51. while (stream->packets.size) {
  52. struct encoder_packet packet;
  53. circlebuf_pop_front(&stream->packets, &packet, sizeof(packet));
  54. obs_encoder_packet_release(&packet);
  55. }
  56. pthread_mutex_unlock(&stream->packets_mutex);
  57. }
  58. static inline bool stopping(struct rtmp_stream *stream)
  59. {
  60. return os_event_try(stream->stop_event) != EAGAIN;
  61. }
  62. static inline bool connecting(struct rtmp_stream *stream)
  63. {
  64. return os_atomic_load_bool(&stream->connecting);
  65. }
  66. static inline bool active(struct rtmp_stream *stream)
  67. {
  68. return os_atomic_load_bool(&stream->active);
  69. }
  70. static inline bool disconnected(struct rtmp_stream *stream)
  71. {
  72. return os_atomic_load_bool(&stream->disconnected);
  73. }
  74. static inline bool silently_reconnecting(struct rtmp_stream *stream)
  75. {
  76. return os_atomic_load_bool(&stream->silent_reconnect);
  77. }
  78. static void rtmp_stream_destroy(void *data)
  79. {
  80. struct rtmp_stream *stream = data;
  81. if (stopping(stream) && !connecting(stream)) {
  82. pthread_join(stream->send_thread, NULL);
  83. } else if (connecting(stream) || active(stream)) {
  84. if (stream->connecting)
  85. pthread_join(stream->connect_thread, NULL);
  86. stream->stop_ts = 0;
  87. os_event_signal(stream->stop_event);
  88. if (active(stream)) {
  89. os_sem_post(stream->send_sem);
  90. obs_output_end_data_capture(stream->output);
  91. pthread_join(stream->send_thread, NULL);
  92. }
  93. }
  94. RTMP_TLS_Free(&stream->rtmp);
  95. free_packets(stream);
  96. dstr_free(&stream->path);
  97. dstr_free(&stream->key);
  98. dstr_free(&stream->username);
  99. dstr_free(&stream->password);
  100. dstr_free(&stream->encoder_name);
  101. dstr_free(&stream->bind_ip);
  102. os_event_destroy(stream->stop_event);
  103. os_sem_destroy(stream->send_sem);
  104. pthread_mutex_destroy(&stream->packets_mutex);
  105. circlebuf_free(&stream->packets);
  106. #ifdef TEST_FRAMEDROPS
  107. circlebuf_free(&stream->droptest_info);
  108. #endif
  109. circlebuf_free(&stream->dbr_frames);
  110. pthread_mutex_destroy(&stream->dbr_mutex);
  111. os_event_destroy(stream->buffer_space_available_event);
  112. os_event_destroy(stream->buffer_has_data_event);
  113. os_event_destroy(stream->socket_available_event);
  114. os_event_destroy(stream->send_thread_signaled_exit);
  115. pthread_mutex_destroy(&stream->write_buf_mutex);
  116. if (stream->write_buf)
  117. bfree(stream->write_buf);
  118. bfree(stream);
  119. }
  120. static void *rtmp_stream_create(obs_data_t *settings, obs_output_t *output)
  121. {
  122. struct rtmp_stream *stream = bzalloc(sizeof(struct rtmp_stream));
  123. stream->output = output;
  124. pthread_mutex_init_value(&stream->packets_mutex);
  125. RTMP_LogSetCallback(log_rtmp);
  126. RTMP_Init(&stream->rtmp);
  127. RTMP_LogSetLevel(RTMP_LOGWARNING);
  128. if (pthread_mutex_init(&stream->packets_mutex, NULL) != 0)
  129. goto fail;
  130. if (os_event_init(&stream->stop_event, OS_EVENT_TYPE_MANUAL) != 0)
  131. goto fail;
  132. if (pthread_mutex_init(&stream->write_buf_mutex, NULL) != 0) {
  133. warn("Failed to initialize write buffer mutex");
  134. goto fail;
  135. }
  136. if (pthread_mutex_init(&stream->dbr_mutex, NULL) != 0) {
  137. warn("Failed to initialize dbr mutex");
  138. goto fail;
  139. }
  140. if (os_event_init(&stream->buffer_space_available_event,
  141. OS_EVENT_TYPE_AUTO) != 0) {
  142. warn("Failed to initialize write buffer event");
  143. goto fail;
  144. }
  145. if (os_event_init(&stream->buffer_has_data_event, OS_EVENT_TYPE_AUTO) !=
  146. 0) {
  147. warn("Failed to initialize data buffer event");
  148. goto fail;
  149. }
  150. if (os_event_init(&stream->socket_available_event,
  151. OS_EVENT_TYPE_AUTO) != 0) {
  152. warn("Failed to initialize socket buffer event");
  153. goto fail;
  154. }
  155. if (os_event_init(&stream->send_thread_signaled_exit,
  156. OS_EVENT_TYPE_MANUAL) != 0) {
  157. warn("Failed to initialize socket exit event");
  158. goto fail;
  159. }
  160. UNUSED_PARAMETER(settings);
  161. return stream;
  162. fail:
  163. rtmp_stream_destroy(stream);
  164. return NULL;
  165. }
  166. static void rtmp_stream_stop(void *data, uint64_t ts)
  167. {
  168. struct rtmp_stream *stream = data;
  169. if (stopping(stream) && ts != 0)
  170. return;
  171. if (connecting(stream))
  172. pthread_join(stream->connect_thread, NULL);
  173. stream->stop_ts = ts / 1000ULL;
  174. if (ts)
  175. stream->shutdown_timeout_ts =
  176. ts +
  177. (uint64_t)stream->max_shutdown_time_sec * 1000000000ULL;
  178. if (active(stream)) {
  179. os_event_signal(stream->stop_event);
  180. if (stream->stop_ts == 0)
  181. os_sem_post(stream->send_sem);
  182. } else {
  183. obs_output_signal_stop(stream->output, OBS_OUTPUT_SUCCESS);
  184. }
  185. }
  186. static inline void set_rtmp_str(AVal *val, const char *str)
  187. {
  188. bool valid = (str && *str);
  189. val->av_val = valid ? (char *)str : NULL;
  190. val->av_len = valid ? (int)strlen(str) : 0;
  191. }
  192. static inline void set_rtmp_dstr(AVal *val, struct dstr *str)
  193. {
  194. bool valid = !dstr_is_empty(str);
  195. val->av_val = valid ? str->array : NULL;
  196. val->av_len = valid ? (int)str->len : 0;
  197. }
  198. static inline bool get_next_packet(struct rtmp_stream *stream,
  199. struct encoder_packet *packet)
  200. {
  201. bool new_packet = false;
  202. pthread_mutex_lock(&stream->packets_mutex);
  203. if (stream->packets.size) {
  204. circlebuf_pop_front(&stream->packets, packet,
  205. sizeof(struct encoder_packet));
  206. new_packet = true;
  207. }
  208. pthread_mutex_unlock(&stream->packets_mutex);
  209. return new_packet;
  210. }
  211. static inline void peek_next_packet(struct rtmp_stream *stream,
  212. struct encoder_packet *packet)
  213. {
  214. pthread_mutex_lock(&stream->packets_mutex);
  215. circlebuf_peek_front(&stream->packets, packet,
  216. sizeof(struct encoder_packet));
  217. pthread_mutex_unlock(&stream->packets_mutex);
  218. }
  219. static void reinsert_packet_at_front(struct rtmp_stream *stream,
  220. struct encoder_packet *packet)
  221. {
  222. pthread_mutex_lock(&stream->packets_mutex);
  223. circlebuf_push_front(&stream->packets, packet,
  224. sizeof(struct encoder_packet));
  225. pthread_mutex_unlock(&stream->packets_mutex);
  226. os_sem_post(stream->send_sem);
  227. }
  228. #define RTMP_PACKET_TYPE_RECONNECT 0x20
  229. static bool process_recv_data(struct rtmp_stream *stream, size_t size)
  230. {
  231. RTMP *rtmp = &stream->rtmp;
  232. RTMPPacket packet = {0};
  233. if (!RTMP_ReadPacket(rtmp, &packet)) {
  234. #ifdef _WIN32
  235. int error = WSAGetLastError();
  236. #else
  237. int error = errno;
  238. #endif
  239. do_log(LOG_ERROR, "RTMP_ReadPacket error: %d", error);
  240. return false;
  241. }
  242. if (packet.m_body) {
  243. if (packet.m_packetType == RTMP_PACKET_TYPE_RECONNECT) {
  244. os_atomic_set_bool(&stream->silent_reconnect, true);
  245. }
  246. RTMPPacket_Free(&packet);
  247. }
  248. return true;
  249. }
  250. #ifdef TEST_FRAMEDROPS
  251. static void droptest_cap_data_rate(struct rtmp_stream *stream, size_t size)
  252. {
  253. uint64_t ts = os_gettime_ns();
  254. struct droptest_info info;
  255. #if defined(_WIN32) && defined(TEST_FRAMEDROPS_WITH_BITRATE_SHORTCUTS)
  256. uint64_t check_elapsed = ts - stream->droptest_last_key_check;
  257. if (check_elapsed > (200ULL * MSEC_TO_NSEC)) {
  258. size_t bitrate = 0;
  259. stream->droptest_last_key_check = ts;
  260. if (GetAsyncKeyState(VK_NUMPAD0) & 0x8000) {
  261. stream->droptest_max = 0;
  262. } else if (GetAsyncKeyState(VK_NUMPAD1) & 0x8000) {
  263. bitrate = 1000;
  264. } else if (GetAsyncKeyState(VK_NUMPAD2) & 0x8000) {
  265. bitrate = 2000;
  266. } else if (GetAsyncKeyState(VK_NUMPAD3) & 0x8000) {
  267. bitrate = 3000;
  268. } else if (GetAsyncKeyState(VK_NUMPAD4) & 0x8000) {
  269. bitrate = 4000;
  270. } else if (GetAsyncKeyState(VK_NUMPAD5) & 0x8000) {
  271. bitrate = 5000;
  272. } else if (GetAsyncKeyState(VK_NUMPAD6) & 0x8000) {
  273. bitrate = 6000;
  274. } else if (GetAsyncKeyState(VK_NUMPAD7) & 0x8000) {
  275. bitrate = 7000;
  276. } else if (GetAsyncKeyState(VK_NUMPAD8) & 0x8000) {
  277. bitrate = 8000;
  278. } else if (GetAsyncKeyState(VK_NUMPAD9) & 0x8000) {
  279. bitrate = 9000;
  280. }
  281. if (bitrate) {
  282. stream->droptest_max = (bitrate * 1000 / 8);
  283. }
  284. }
  285. if (!stream->droptest_max) {
  286. return;
  287. }
  288. #else
  289. if (!stream->droptest_max) {
  290. stream->droptest_max = DROPTEST_MAX_BYTES;
  291. }
  292. #endif
  293. info.ts = ts;
  294. info.size = size;
  295. circlebuf_push_back(&stream->droptest_info, &info, sizeof(info));
  296. stream->droptest_size += size;
  297. if (stream->droptest_info.size) {
  298. circlebuf_peek_front(&stream->droptest_info, &info,
  299. sizeof(info));
  300. if (stream->droptest_size > stream->droptest_max) {
  301. uint64_t elapsed = ts - info.ts;
  302. if (elapsed < 1000000000ULL) {
  303. elapsed = 1000000000ULL - elapsed;
  304. os_sleepto_ns(ts + elapsed);
  305. }
  306. while (stream->droptest_size > stream->droptest_max) {
  307. circlebuf_pop_front(&stream->droptest_info,
  308. &info, sizeof(info));
  309. stream->droptest_size -= info.size;
  310. }
  311. }
  312. }
  313. }
  314. #endif
  315. static int socket_queue_data(RTMPSockBuf *sb, const char *data, int len,
  316. void *arg)
  317. {
  318. UNUSED_PARAMETER(sb);
  319. struct rtmp_stream *stream = arg;
  320. retry_send:
  321. if (!RTMP_IsConnected(&stream->rtmp))
  322. return 0;
  323. pthread_mutex_lock(&stream->write_buf_mutex);
  324. if (stream->write_buf_len + len > stream->write_buf_size) {
  325. pthread_mutex_unlock(&stream->write_buf_mutex);
  326. if (os_event_wait(stream->buffer_space_available_event)) {
  327. return 0;
  328. }
  329. goto retry_send;
  330. }
  331. memcpy(stream->write_buf + stream->write_buf_len, data, len);
  332. stream->write_buf_len += len;
  333. pthread_mutex_unlock(&stream->write_buf_mutex);
  334. os_event_signal(stream->buffer_has_data_event);
  335. return len;
  336. }
  337. static int send_packet(struct rtmp_stream *stream,
  338. struct encoder_packet *packet, bool is_header,
  339. size_t idx)
  340. {
  341. uint8_t *data;
  342. size_t size;
  343. int recv_size = 0;
  344. int ret = 0;
  345. assert(idx < RTMP_MAX_STREAMS);
  346. if (!stream->new_socket_loop) {
  347. #ifdef _WIN32
  348. ret = ioctlsocket(stream->rtmp.m_sb.sb_socket, FIONREAD,
  349. (u_long *)&recv_size);
  350. #else
  351. ret = ioctl(stream->rtmp.m_sb.sb_socket, FIONREAD, &recv_size);
  352. #endif
  353. if (ret >= 0 && recv_size > 0) {
  354. if (!process_recv_data(stream, (size_t)recv_size))
  355. return -1;
  356. }
  357. }
  358. if (idx > 0) {
  359. flv_additional_packet_mux(
  360. packet, is_header ? 0 : stream->start_dts_offset, &data,
  361. &size, is_header, idx);
  362. } else {
  363. flv_packet_mux(packet, is_header ? 0 : stream->start_dts_offset,
  364. &data, &size, is_header);
  365. }
  366. #ifdef TEST_FRAMEDROPS
  367. droptest_cap_data_rate(stream, size);
  368. #endif
  369. ret = RTMP_Write(&stream->rtmp, (char *)data, (int)size, 0);
  370. bfree(data);
  371. if (is_header)
  372. bfree(packet->data);
  373. else
  374. obs_encoder_packet_release(packet);
  375. stream->total_bytes_sent += size;
  376. return ret;
  377. }
  378. static inline bool send_headers(struct rtmp_stream *stream);
  379. static inline bool can_shutdown_stream(struct rtmp_stream *stream,
  380. struct encoder_packet *packet)
  381. {
  382. uint64_t cur_time = os_gettime_ns();
  383. bool timeout = cur_time >= stream->shutdown_timeout_ts;
  384. if (timeout)
  385. info("Stream shutdown timeout reached (%d second(s))",
  386. stream->max_shutdown_time_sec);
  387. return timeout || packet->sys_dts_usec >= (int64_t)stream->stop_ts;
  388. }
  389. static void set_output_error(struct rtmp_stream *stream)
  390. {
  391. const char *msg = NULL;
  392. #ifdef _WIN32
  393. switch (stream->rtmp.last_error_code) {
  394. case WSAETIMEDOUT:
  395. msg = obs_module_text("ConnectionTimedOut");
  396. break;
  397. case WSAEACCES:
  398. msg = obs_module_text("PermissionDenied");
  399. break;
  400. case WSAECONNABORTED:
  401. msg = obs_module_text("ConnectionAborted");
  402. break;
  403. case WSAECONNRESET:
  404. msg = obs_module_text("ConnectionReset");
  405. break;
  406. case WSAHOST_NOT_FOUND:
  407. msg = obs_module_text("HostNotFound");
  408. break;
  409. case WSANO_DATA:
  410. msg = obs_module_text("NoData");
  411. break;
  412. case WSAEADDRNOTAVAIL:
  413. msg = obs_module_text("AddressNotAvailable");
  414. break;
  415. }
  416. #else
  417. switch (stream->rtmp.last_error_code) {
  418. case ETIMEDOUT:
  419. msg = obs_module_text("ConnectionTimedOut");
  420. break;
  421. case EACCES:
  422. msg = obs_module_text("PermissionDenied");
  423. break;
  424. case ECONNABORTED:
  425. msg = obs_module_text("ConnectionAborted");
  426. break;
  427. case ECONNRESET:
  428. msg = obs_module_text("ConnectionReset");
  429. break;
  430. case HOST_NOT_FOUND:
  431. msg = obs_module_text("HostNotFound");
  432. break;
  433. case NO_DATA:
  434. msg = obs_module_text("NoData");
  435. break;
  436. case EADDRNOTAVAIL:
  437. msg = obs_module_text("AddressNotAvailable");
  438. break;
  439. }
  440. #endif
  441. // non platform-specific errors
  442. if (!msg) {
  443. switch (stream->rtmp.last_error_code) {
  444. case -0x2700:
  445. msg = obs_module_text("SSLCertVerifyFailed");
  446. break;
  447. case -0x7680:
  448. msg = "Failed to load root certificates for a secure TLS connection."
  449. #if defined(__linux__)
  450. " Check you have an up to date root certificate bundle in /etc/ssl/certs."
  451. #endif
  452. ;
  453. break;
  454. }
  455. }
  456. if (msg)
  457. obs_output_set_last_error(stream->output, msg);
  458. }
  459. static void dbr_add_frame(struct rtmp_stream *stream, struct dbr_frame *back)
  460. {
  461. struct dbr_frame front;
  462. uint64_t dur;
  463. circlebuf_push_back(&stream->dbr_frames, back, sizeof(*back));
  464. circlebuf_peek_front(&stream->dbr_frames, &front, sizeof(front));
  465. stream->dbr_data_size += back->size;
  466. dur = (back->send_end - front.send_beg) / 1000000;
  467. if (dur >= MAX_ESTIMATE_DURATION_MS) {
  468. stream->dbr_data_size -= front.size;
  469. circlebuf_pop_front(&stream->dbr_frames, NULL, sizeof(front));
  470. }
  471. stream->dbr_est_bitrate =
  472. (dur >= MIN_ESTIMATE_DURATION_MS)
  473. ? (long)(stream->dbr_data_size * 1000 / dur)
  474. : 0;
  475. stream->dbr_est_bitrate *= 8;
  476. stream->dbr_est_bitrate /= 1000;
  477. if (stream->dbr_est_bitrate) {
  478. stream->dbr_est_bitrate -= stream->audio_bitrate;
  479. if (stream->dbr_est_bitrate < 50)
  480. stream->dbr_est_bitrate = 50;
  481. }
  482. }
  483. static void dbr_set_bitrate(struct rtmp_stream *stream);
  484. static bool rtmp_stream_start(void *data);
  485. #ifdef _WIN32
  486. #define socklen_t int
  487. #endif
  488. static void log_sndbuf_size(struct rtmp_stream *stream)
  489. {
  490. int cur_sendbuf_size;
  491. socklen_t int_size = sizeof(int);
  492. if (!getsockopt(stream->rtmp.m_sb.sb_socket, SOL_SOCKET, SO_SNDBUF,
  493. (char *)&cur_sendbuf_size, &int_size)) {
  494. info("Socket send buffer is %d bytes", cur_sendbuf_size);
  495. }
  496. }
  497. static void *send_thread(void *data)
  498. {
  499. struct rtmp_stream *stream = data;
  500. os_set_thread_name("rtmp-stream: send_thread");
  501. #if defined(_WIN32)
  502. // Despite MSDN claiming otherwise, send buffer auto tuning on
  503. // Windows 7 doesn't seem to work very well.
  504. if (get_win_ver_int() == 0x601) {
  505. DWORD cur_sendbuf_size;
  506. DWORD desired_sendbuf_size = 524288;
  507. socklen_t int_size = sizeof(int);
  508. if (!getsockopt(stream->rtmp.m_sb.sb_socket, SOL_SOCKET,
  509. SO_SNDBUF, (char *)&cur_sendbuf_size,
  510. &int_size) &&
  511. cur_sendbuf_size < desired_sendbuf_size) {
  512. setsockopt(stream->rtmp.m_sb.sb_socket, SOL_SOCKET,
  513. SO_SNDBUF, (char *)&desired_sendbuf_size,
  514. sizeof(desired_sendbuf_size));
  515. }
  516. }
  517. log_sndbuf_size(stream);
  518. #endif
  519. while (os_sem_wait(stream->send_sem) == 0) {
  520. struct encoder_packet packet;
  521. struct dbr_frame dbr_frame;
  522. if (stopping(stream) && stream->stop_ts == 0) {
  523. break;
  524. }
  525. if (!get_next_packet(stream, &packet))
  526. continue;
  527. if (stopping(stream)) {
  528. if (can_shutdown_stream(stream, &packet)) {
  529. obs_encoder_packet_release(&packet);
  530. break;
  531. }
  532. }
  533. if (!stream->sent_headers) {
  534. if (!send_headers(stream)) {
  535. os_atomic_set_bool(&stream->disconnected, true);
  536. break;
  537. }
  538. }
  539. /* silent reconnect signal received from server, reconnect on
  540. * next keyframe */
  541. if (silently_reconnecting(stream) &&
  542. packet.type == OBS_ENCODER_VIDEO && packet.keyframe) {
  543. reinsert_packet_at_front(stream, &packet);
  544. break;
  545. }
  546. if (stream->dbr_enabled) {
  547. dbr_frame.send_beg = os_gettime_ns();
  548. dbr_frame.size = packet.size;
  549. }
  550. if (send_packet(stream, &packet, false, packet.track_idx) < 0) {
  551. os_atomic_set_bool(&stream->disconnected, true);
  552. break;
  553. }
  554. if (stream->dbr_enabled) {
  555. dbr_frame.send_end = os_gettime_ns();
  556. pthread_mutex_lock(&stream->dbr_mutex);
  557. dbr_add_frame(stream, &dbr_frame);
  558. pthread_mutex_unlock(&stream->dbr_mutex);
  559. }
  560. }
  561. bool encode_error = os_atomic_load_bool(&stream->encode_error);
  562. if (disconnected(stream)) {
  563. info("Disconnected from %s", stream->path.array);
  564. } else if (encode_error) {
  565. info("Encoder error, disconnecting");
  566. } else if (silently_reconnecting(stream)) {
  567. info("Silent reconnect signal received from server");
  568. } else {
  569. info("User stopped the stream");
  570. }
  571. #if defined(_WIN32)
  572. log_sndbuf_size(stream);
  573. #endif
  574. if (stream->new_socket_loop) {
  575. os_event_signal(stream->send_thread_signaled_exit);
  576. os_event_signal(stream->buffer_has_data_event);
  577. pthread_join(stream->socket_thread, NULL);
  578. stream->socket_thread_active = false;
  579. stream->rtmp.m_bCustomSend = false;
  580. }
  581. set_output_error(stream);
  582. RTMP_Close(&stream->rtmp);
  583. /* reset bitrate on stop */
  584. if (stream->dbr_enabled) {
  585. if (stream->dbr_cur_bitrate != stream->dbr_orig_bitrate) {
  586. stream->dbr_cur_bitrate = stream->dbr_orig_bitrate;
  587. dbr_set_bitrate(stream);
  588. }
  589. }
  590. if (!stopping(stream)) {
  591. pthread_detach(stream->send_thread);
  592. if (!silently_reconnecting(stream))
  593. obs_output_signal_stop(stream->output,
  594. OBS_OUTPUT_DISCONNECTED);
  595. } else if (encode_error) {
  596. obs_output_signal_stop(stream->output, OBS_OUTPUT_ENCODE_ERROR);
  597. } else {
  598. obs_output_end_data_capture(stream->output);
  599. }
  600. if (!silently_reconnecting(stream)) {
  601. free_packets(stream);
  602. os_event_reset(stream->stop_event);
  603. os_atomic_set_bool(&stream->active, false);
  604. }
  605. stream->sent_headers = false;
  606. /* reset bitrate on stop */
  607. if (stream->dbr_enabled) {
  608. if (stream->dbr_cur_bitrate != stream->dbr_orig_bitrate) {
  609. stream->dbr_cur_bitrate = stream->dbr_orig_bitrate;
  610. dbr_set_bitrate(stream);
  611. }
  612. }
  613. if (silently_reconnecting(stream)) {
  614. rtmp_stream_start(stream);
  615. }
  616. return NULL;
  617. }
  618. static bool send_additional_meta_data(struct rtmp_stream *stream)
  619. {
  620. uint8_t *meta_data;
  621. size_t meta_data_size;
  622. bool success = true;
  623. flv_additional_meta_data(stream->output, &meta_data, &meta_data_size);
  624. success = RTMP_Write(&stream->rtmp, (char *)meta_data,
  625. (int)meta_data_size, 0) >= 0;
  626. bfree(meta_data);
  627. return success;
  628. }
  629. static bool send_meta_data(struct rtmp_stream *stream)
  630. {
  631. uint8_t *meta_data;
  632. size_t meta_data_size;
  633. bool success = true;
  634. flv_meta_data(stream->output, &meta_data, &meta_data_size, false);
  635. success = RTMP_Write(&stream->rtmp, (char *)meta_data,
  636. (int)meta_data_size, 0) >= 0;
  637. bfree(meta_data);
  638. return success;
  639. }
  640. static bool send_audio_header(struct rtmp_stream *stream, size_t idx,
  641. bool *next)
  642. {
  643. obs_output_t *context = stream->output;
  644. obs_encoder_t *aencoder = obs_output_get_audio_encoder(context, idx);
  645. uint8_t *header;
  646. struct encoder_packet packet = {.type = OBS_ENCODER_AUDIO,
  647. .timebase_den = 1};
  648. if (!aencoder) {
  649. *next = false;
  650. return true;
  651. }
  652. obs_encoder_get_extra_data(aencoder, &header, &packet.size);
  653. packet.data = bmemdup(header, packet.size);
  654. return send_packet(stream, &packet, true, idx) >= 0;
  655. }
  656. static bool send_video_header(struct rtmp_stream *stream)
  657. {
  658. obs_output_t *context = stream->output;
  659. obs_encoder_t *vencoder = obs_output_get_video_encoder(context);
  660. uint8_t *header;
  661. size_t size;
  662. struct encoder_packet packet = {
  663. .type = OBS_ENCODER_VIDEO, .timebase_den = 1, .keyframe = true};
  664. obs_encoder_get_extra_data(vencoder, &header, &size);
  665. packet.size = obs_parse_avc_header(&packet.data, header, size);
  666. return send_packet(stream, &packet, true, 0) >= 0;
  667. }
  668. static inline bool send_headers(struct rtmp_stream *stream)
  669. {
  670. stream->sent_headers = true;
  671. size_t i = 0;
  672. bool next = true;
  673. if (!send_audio_header(stream, i++, &next))
  674. return false;
  675. if (!send_video_header(stream))
  676. return false;
  677. while (next) {
  678. if (!send_audio_header(stream, i++, &next))
  679. return false;
  680. }
  681. return true;
  682. }
  683. static inline bool reset_semaphore(struct rtmp_stream *stream)
  684. {
  685. os_sem_destroy(stream->send_sem);
  686. return os_sem_init(&stream->send_sem, 0) == 0;
  687. }
  688. static int init_send(struct rtmp_stream *stream)
  689. {
  690. int ret;
  691. obs_output_t *context = stream->output;
  692. if (!silently_reconnecting(stream))
  693. reset_semaphore(stream);
  694. ret = pthread_create(&stream->send_thread, NULL, send_thread, stream);
  695. if (ret != 0) {
  696. RTMP_Close(&stream->rtmp);
  697. warn("Failed to create send thread");
  698. return OBS_OUTPUT_ERROR;
  699. }
  700. if (stream->new_socket_loop) {
  701. int one = 1;
  702. #ifdef _WIN32
  703. if (ioctlsocket(stream->rtmp.m_sb.sb_socket, FIONBIO, &one)) {
  704. stream->rtmp.last_error_code = WSAGetLastError();
  705. #else
  706. if (ioctl(stream->rtmp.m_sb.sb_socket, FIONBIO, &one)) {
  707. stream->rtmp.last_error_code = errno;
  708. #endif
  709. warn("Failed to set non-blocking socket");
  710. return OBS_OUTPUT_ERROR;
  711. }
  712. os_event_reset(stream->send_thread_signaled_exit);
  713. info("New socket loop enabled by user");
  714. if (stream->low_latency_mode)
  715. info("Low latency mode enabled by user");
  716. if (stream->write_buf)
  717. bfree(stream->write_buf);
  718. int total_bitrate = 0;
  719. obs_encoder_t *vencoder = obs_output_get_video_encoder(context);
  720. if (vencoder) {
  721. obs_data_t *params = obs_encoder_get_settings(vencoder);
  722. if (params) {
  723. int bitrate =
  724. obs_data_get_int(params, "bitrate");
  725. if (!bitrate) {
  726. warn("Video encoder didn't return a "
  727. "valid bitrate, new network "
  728. "code may function poorly. "
  729. "Low latency mode disabled.");
  730. stream->low_latency_mode = false;
  731. bitrate = 10000;
  732. }
  733. total_bitrate += bitrate;
  734. obs_data_release(params);
  735. }
  736. }
  737. obs_encoder_t *aencoder =
  738. obs_output_get_audio_encoder(context, 0);
  739. if (aencoder) {
  740. obs_data_t *params = obs_encoder_get_settings(aencoder);
  741. if (params) {
  742. int bitrate =
  743. obs_data_get_int(params, "bitrate");
  744. if (!bitrate)
  745. bitrate = 160;
  746. total_bitrate += bitrate;
  747. obs_data_release(params);
  748. }
  749. }
  750. // to bytes/sec
  751. int ideal_buffer_size = total_bitrate * 128;
  752. if (ideal_buffer_size < 131072)
  753. ideal_buffer_size = 131072;
  754. stream->write_buf_size = ideal_buffer_size;
  755. stream->write_buf = bmalloc(ideal_buffer_size);
  756. #ifdef _WIN32
  757. ret = pthread_create(&stream->socket_thread, NULL,
  758. socket_thread_windows, stream);
  759. #else
  760. warn("New socket loop not supported on this platform");
  761. return OBS_OUTPUT_ERROR;
  762. #endif
  763. if (ret != 0) {
  764. RTMP_Close(&stream->rtmp);
  765. warn("Failed to create socket thread");
  766. return OBS_OUTPUT_ERROR;
  767. }
  768. stream->socket_thread_active = true;
  769. stream->rtmp.m_bCustomSend = true;
  770. stream->rtmp.m_customSendFunc = socket_queue_data;
  771. stream->rtmp.m_customSendParam = stream;
  772. }
  773. os_atomic_set_bool(&stream->active, true);
  774. if (!send_meta_data(stream)) {
  775. warn("Disconnected while attempting to send metadata");
  776. set_output_error(stream);
  777. return OBS_OUTPUT_DISCONNECTED;
  778. }
  779. obs_encoder_t *aencoder = obs_output_get_audio_encoder(context, 1);
  780. if (aencoder && !send_additional_meta_data(stream)) {
  781. warn("Disconnected while attempting to send additional "
  782. "metadata");
  783. return OBS_OUTPUT_DISCONNECTED;
  784. }
  785. if (obs_output_get_audio_encoder(context, 2) != NULL) {
  786. warn("Additional audio streams not supported");
  787. return OBS_OUTPUT_DISCONNECTED;
  788. }
  789. if (!silently_reconnecting(stream))
  790. obs_output_begin_data_capture(stream->output, 0);
  791. return OBS_OUTPUT_SUCCESS;
  792. }
  793. #ifdef _WIN32
  794. static void win32_log_interface_type(struct rtmp_stream *stream)
  795. {
  796. RTMP *rtmp = &stream->rtmp;
  797. MIB_IPFORWARDROW route;
  798. uint32_t dest_addr, source_addr;
  799. char hostname[256];
  800. HOSTENT *h;
  801. if (rtmp->Link.hostname.av_len >= sizeof(hostname) - 1)
  802. return;
  803. strncpy(hostname, rtmp->Link.hostname.av_val, sizeof(hostname));
  804. hostname[rtmp->Link.hostname.av_len] = 0;
  805. h = gethostbyname(hostname);
  806. if (!h)
  807. return;
  808. dest_addr = *(uint32_t *)h->h_addr_list[0];
  809. if (rtmp->m_bindIP.addrLen == 0)
  810. source_addr = 0;
  811. else if (rtmp->m_bindIP.addr.ss_family == AF_INET)
  812. source_addr = (*(struct sockaddr_in *)&rtmp->m_bindIP.addr)
  813. .sin_addr.S_un.S_addr;
  814. else
  815. return;
  816. if (!GetBestRoute(dest_addr, source_addr, &route)) {
  817. MIB_IFROW row;
  818. memset(&row, 0, sizeof(row));
  819. row.dwIndex = route.dwForwardIfIndex;
  820. if (!GetIfEntry(&row)) {
  821. uint32_t speed = row.dwSpeed / 1000000;
  822. char *type;
  823. struct dstr other = {0};
  824. if (row.dwType == IF_TYPE_ETHERNET_CSMACD) {
  825. type = "ethernet";
  826. } else if (row.dwType == IF_TYPE_IEEE80211) {
  827. type = "802.11";
  828. } else {
  829. dstr_printf(&other, "type %lu", row.dwType);
  830. type = other.array;
  831. }
  832. info("Interface: %s (%s, %lu mbps)", row.bDescr, type,
  833. speed);
  834. dstr_free(&other);
  835. }
  836. }
  837. }
  838. #endif
  839. static void add_connect_data(char **penc, char *pend)
  840. {
  841. const AVal val = AVC("supportsGoAway");
  842. *penc = AMF_EncodeNamedBoolean(*penc, pend, &val, true);
  843. }
  844. static int try_connect(struct rtmp_stream *stream)
  845. {
  846. if (dstr_is_empty(&stream->path)) {
  847. warn("URL is empty");
  848. return OBS_OUTPUT_BAD_PATH;
  849. }
  850. info("Connecting to RTMP URL %s...", stream->path.array);
  851. // on reconnect we need to reset the internal variables of librtmp
  852. // otherwise the data sent/received will not parse correctly on the other end
  853. RTMP_Reset(&stream->rtmp);
  854. // since we don't call RTMP_Init above, there's no other good place
  855. // to reset this as doing it in RTMP_Close breaks the ugly RTMP
  856. // authentication system
  857. memset(&stream->rtmp.Link, 0, sizeof(stream->rtmp.Link));
  858. stream->rtmp.last_error_code = 0;
  859. if (!RTMP_SetupURL(&stream->rtmp, stream->path.array))
  860. return OBS_OUTPUT_BAD_PATH;
  861. RTMP_EnableWrite(&stream->rtmp);
  862. dstr_copy(&stream->encoder_name, "FMLE/3.0 (compatible; FMSc/1.0)");
  863. set_rtmp_dstr(&stream->rtmp.Link.pubUser, &stream->username);
  864. set_rtmp_dstr(&stream->rtmp.Link.pubPasswd, &stream->password);
  865. set_rtmp_dstr(&stream->rtmp.Link.flashVer, &stream->encoder_name);
  866. stream->rtmp.Link.swfUrl = stream->rtmp.Link.tcUrl;
  867. stream->rtmp.Link.customConnectEncode = add_connect_data;
  868. if (dstr_is_empty(&stream->bind_ip) ||
  869. dstr_cmp(&stream->bind_ip, "default") == 0) {
  870. memset(&stream->rtmp.m_bindIP, 0,
  871. sizeof(stream->rtmp.m_bindIP));
  872. } else {
  873. bool success = netif_str_to_addr(&stream->rtmp.m_bindIP.addr,
  874. &stream->rtmp.m_bindIP.addrLen,
  875. stream->bind_ip.array);
  876. if (success) {
  877. int len = stream->rtmp.m_bindIP.addrLen;
  878. bool ipv6 = len == sizeof(struct sockaddr_in6);
  879. info("Binding to IPv%d", ipv6 ? 6 : 4);
  880. }
  881. }
  882. RTMP_AddStream(&stream->rtmp, stream->key.array);
  883. stream->rtmp.m_outChunkSize = 4096;
  884. stream->rtmp.m_bSendChunkSizeInfo = true;
  885. stream->rtmp.m_bUseNagle = true;
  886. #ifdef _WIN32
  887. win32_log_interface_type(stream);
  888. #endif
  889. if (!RTMP_Connect(&stream->rtmp, NULL)) {
  890. set_output_error(stream);
  891. return OBS_OUTPUT_CONNECT_FAILED;
  892. }
  893. if (!RTMP_ConnectStream(&stream->rtmp, 0))
  894. return OBS_OUTPUT_INVALID_STREAM;
  895. info("Connection to %s successful", stream->path.array);
  896. return init_send(stream);
  897. }
  898. static bool init_connect(struct rtmp_stream *stream)
  899. {
  900. obs_service_t *service;
  901. obs_data_t *settings;
  902. const char *bind_ip;
  903. int64_t drop_p;
  904. int64_t drop_b;
  905. uint32_t caps;
  906. if (stopping(stream)) {
  907. pthread_join(stream->send_thread, NULL);
  908. }
  909. free_packets(stream);
  910. service = obs_output_get_service(stream->output);
  911. if (!service)
  912. return false;
  913. os_atomic_set_bool(&stream->disconnected, false);
  914. os_atomic_set_bool(&stream->encode_error, false);
  915. stream->total_bytes_sent = 0;
  916. stream->dropped_frames = 0;
  917. stream->min_priority = 0;
  918. stream->got_first_video = false;
  919. settings = obs_output_get_settings(stream->output);
  920. dstr_copy(&stream->path, obs_service_get_url(service));
  921. dstr_copy(&stream->key, obs_service_get_key(service));
  922. dstr_copy(&stream->username, obs_service_get_username(service));
  923. dstr_copy(&stream->password, obs_service_get_password(service));
  924. dstr_depad(&stream->path);
  925. dstr_depad(&stream->key);
  926. drop_b = (int64_t)obs_data_get_int(settings, OPT_DROP_THRESHOLD);
  927. drop_p = (int64_t)obs_data_get_int(settings, OPT_PFRAME_DROP_THRESHOLD);
  928. stream->max_shutdown_time_sec =
  929. (int)obs_data_get_int(settings, OPT_MAX_SHUTDOWN_TIME_SEC);
  930. obs_encoder_t *venc = obs_output_get_video_encoder(stream->output);
  931. obs_encoder_t *aenc = obs_output_get_audio_encoder(stream->output, 0);
  932. obs_data_t *vsettings = obs_encoder_get_settings(venc);
  933. obs_data_t *asettings = obs_encoder_get_settings(aenc);
  934. circlebuf_free(&stream->dbr_frames);
  935. stream->audio_bitrate = (long)obs_data_get_int(asettings, "bitrate");
  936. stream->dbr_data_size = 0;
  937. stream->dbr_orig_bitrate = (long)obs_data_get_int(vsettings, "bitrate");
  938. stream->dbr_cur_bitrate = stream->dbr_orig_bitrate;
  939. stream->dbr_est_bitrate = 0;
  940. stream->dbr_inc_bitrate = stream->dbr_orig_bitrate / 10;
  941. stream->dbr_inc_timeout = 0;
  942. stream->dbr_enabled = obs_data_get_bool(settings, OPT_DYN_BITRATE);
  943. caps = obs_encoder_get_caps(venc);
  944. if ((caps & OBS_ENCODER_CAP_DYN_BITRATE) == 0) {
  945. stream->dbr_enabled = false;
  946. }
  947. if (obs_output_get_delay(stream->output) != 0) {
  948. stream->dbr_enabled = false;
  949. }
  950. if (stream->dbr_enabled) {
  951. info("Dynamic bitrate enabled. Dropped frames begone!");
  952. }
  953. obs_data_release(vsettings);
  954. obs_data_release(asettings);
  955. if (drop_p < (drop_b + 200))
  956. drop_p = drop_b + 200;
  957. stream->drop_threshold_usec = 1000 * drop_b;
  958. stream->pframe_drop_threshold_usec = 1000 * drop_p;
  959. bind_ip = obs_data_get_string(settings, OPT_BIND_IP);
  960. dstr_copy(&stream->bind_ip, bind_ip);
  961. stream->new_socket_loop =
  962. obs_data_get_bool(settings, OPT_NEWSOCKETLOOP_ENABLED);
  963. stream->low_latency_mode =
  964. obs_data_get_bool(settings, OPT_LOWLATENCY_ENABLED);
  965. // ugly hack for now, can be removed once new loop is reworked
  966. if (stream->new_socket_loop &&
  967. !strncmp(stream->path.array, "rtmps://", 8)) {
  968. warn("Disabling network optimizations, not compatible with RTMPS");
  969. stream->new_socket_loop = false;
  970. }
  971. obs_data_release(settings);
  972. return true;
  973. }
  974. static void *connect_thread(void *data)
  975. {
  976. struct rtmp_stream *stream = data;
  977. int ret;
  978. os_set_thread_name("rtmp-stream: connect_thread");
  979. if (!silently_reconnecting(stream)) {
  980. if (!init_connect(stream)) {
  981. obs_output_signal_stop(stream->output,
  982. OBS_OUTPUT_BAD_PATH);
  983. os_atomic_set_bool(&stream->silent_reconnect, false);
  984. return NULL;
  985. }
  986. } else {
  987. struct encoder_packet packet;
  988. peek_next_packet(stream, &packet);
  989. stream->start_dts_offset = get_ms_time(&packet, packet.dts);
  990. }
  991. ret = try_connect(stream);
  992. if (ret != OBS_OUTPUT_SUCCESS) {
  993. obs_output_signal_stop(stream->output, ret);
  994. info("Connection to %s failed: %d", stream->path.array, ret);
  995. }
  996. if (!stopping(stream))
  997. pthread_detach(stream->connect_thread);
  998. os_atomic_set_bool(&stream->silent_reconnect, false);
  999. os_atomic_set_bool(&stream->connecting, false);
  1000. return NULL;
  1001. }
  1002. static bool rtmp_stream_start(void *data)
  1003. {
  1004. struct rtmp_stream *stream = data;
  1005. if (!silently_reconnecting(stream)) {
  1006. if (!obs_output_can_begin_data_capture(stream->output, 0))
  1007. return false;
  1008. if (!obs_output_initialize_encoders(stream->output, 0))
  1009. return false;
  1010. }
  1011. os_atomic_set_bool(&stream->connecting, true);
  1012. return pthread_create(&stream->connect_thread, NULL, connect_thread,
  1013. stream) == 0;
  1014. }
  1015. static inline bool add_packet(struct rtmp_stream *stream,
  1016. struct encoder_packet *packet)
  1017. {
  1018. circlebuf_push_back(&stream->packets, packet,
  1019. sizeof(struct encoder_packet));
  1020. return true;
  1021. }
  1022. static inline size_t num_buffered_packets(struct rtmp_stream *stream)
  1023. {
  1024. return stream->packets.size / sizeof(struct encoder_packet);
  1025. }
  1026. static void drop_frames(struct rtmp_stream *stream, const char *name,
  1027. int highest_priority, bool pframes)
  1028. {
  1029. UNUSED_PARAMETER(pframes);
  1030. struct circlebuf new_buf = {0};
  1031. int num_frames_dropped = 0;
  1032. #ifdef _DEBUG
  1033. int start_packets = (int)num_buffered_packets(stream);
  1034. #else
  1035. UNUSED_PARAMETER(name);
  1036. #endif
  1037. circlebuf_reserve(&new_buf, sizeof(struct encoder_packet) * 8);
  1038. while (stream->packets.size) {
  1039. struct encoder_packet packet;
  1040. circlebuf_pop_front(&stream->packets, &packet, sizeof(packet));
  1041. /* do not drop audio data or video keyframes */
  1042. if (packet.type == OBS_ENCODER_AUDIO ||
  1043. packet.drop_priority >= highest_priority) {
  1044. circlebuf_push_back(&new_buf, &packet, sizeof(packet));
  1045. } else {
  1046. num_frames_dropped++;
  1047. obs_encoder_packet_release(&packet);
  1048. }
  1049. }
  1050. circlebuf_free(&stream->packets);
  1051. stream->packets = new_buf;
  1052. if (stream->min_priority < highest_priority)
  1053. stream->min_priority = highest_priority;
  1054. if (!num_frames_dropped)
  1055. return;
  1056. stream->dropped_frames += num_frames_dropped;
  1057. #ifdef _DEBUG
  1058. debug("Dropped %s, prev packet count: %d, new packet count: %d", name,
  1059. start_packets, (int)num_buffered_packets(stream));
  1060. #endif
  1061. }
  1062. static bool find_first_video_packet(struct rtmp_stream *stream,
  1063. struct encoder_packet *first)
  1064. {
  1065. size_t count = stream->packets.size / sizeof(*first);
  1066. for (size_t i = 0; i < count; i++) {
  1067. struct encoder_packet *cur =
  1068. circlebuf_data(&stream->packets, i * sizeof(*first));
  1069. if (cur->type == OBS_ENCODER_VIDEO && !cur->keyframe) {
  1070. *first = *cur;
  1071. return true;
  1072. }
  1073. }
  1074. return false;
  1075. }
  1076. static bool dbr_bitrate_lowered(struct rtmp_stream *stream)
  1077. {
  1078. long prev_bitrate = stream->dbr_prev_bitrate;
  1079. long est_bitrate = 0;
  1080. long new_bitrate;
  1081. if (stream->dbr_est_bitrate &&
  1082. stream->dbr_est_bitrate < stream->dbr_cur_bitrate) {
  1083. stream->dbr_data_size = 0;
  1084. circlebuf_pop_front(&stream->dbr_frames, NULL,
  1085. stream->dbr_frames.size);
  1086. est_bitrate = stream->dbr_est_bitrate / 100 * 100;
  1087. if (est_bitrate < 50) {
  1088. est_bitrate = 50;
  1089. }
  1090. }
  1091. #if 0
  1092. if (prev_bitrate && est_bitrate) {
  1093. if (prev_bitrate < est_bitrate) {
  1094. blog(LOG_INFO, "going back to prev bitrate: "
  1095. "prev_bitrate (%d) < est_bitrate (%d)",
  1096. prev_bitrate,
  1097. est_bitrate);
  1098. new_bitrate = prev_bitrate;
  1099. } else {
  1100. new_bitrate = est_bitrate;
  1101. }
  1102. new_bitrate = est_bitrate;
  1103. } else if (prev_bitrate) {
  1104. new_bitrate = prev_bitrate;
  1105. info("going back to prev bitrate");
  1106. } else if (est_bitrate) {
  1107. new_bitrate = est_bitrate;
  1108. } else {
  1109. return false;
  1110. }
  1111. #else
  1112. if (est_bitrate) {
  1113. new_bitrate = est_bitrate;
  1114. } else if (prev_bitrate) {
  1115. new_bitrate = prev_bitrate;
  1116. info("going back to prev bitrate");
  1117. } else {
  1118. return false;
  1119. }
  1120. if (new_bitrate == stream->dbr_cur_bitrate) {
  1121. return false;
  1122. }
  1123. #endif
  1124. stream->dbr_prev_bitrate = 0;
  1125. stream->dbr_cur_bitrate = new_bitrate;
  1126. stream->dbr_inc_timeout = os_gettime_ns() + DBR_INC_TIMER;
  1127. info("bitrate decreased to: %ld", stream->dbr_cur_bitrate);
  1128. return true;
  1129. }
  1130. static void dbr_set_bitrate(struct rtmp_stream *stream)
  1131. {
  1132. obs_encoder_t *vencoder = obs_output_get_video_encoder(stream->output);
  1133. obs_data_t *settings = obs_encoder_get_settings(vencoder);
  1134. obs_data_set_int(settings, "bitrate", stream->dbr_cur_bitrate);
  1135. obs_encoder_update(vencoder, settings);
  1136. obs_data_release(settings);
  1137. }
  1138. static void dbr_inc_bitrate(struct rtmp_stream *stream)
  1139. {
  1140. stream->dbr_prev_bitrate = stream->dbr_cur_bitrate;
  1141. stream->dbr_cur_bitrate += stream->dbr_inc_bitrate;
  1142. if (stream->dbr_cur_bitrate >= stream->dbr_orig_bitrate) {
  1143. stream->dbr_cur_bitrate = stream->dbr_orig_bitrate;
  1144. info("bitrate increased to: %ld, done",
  1145. stream->dbr_cur_bitrate);
  1146. } else if (stream->dbr_cur_bitrate < stream->dbr_orig_bitrate) {
  1147. stream->dbr_inc_timeout = os_gettime_ns() + DBR_INC_TIMER;
  1148. info("bitrate increased to: %ld, waiting",
  1149. stream->dbr_cur_bitrate);
  1150. }
  1151. }
  1152. static void check_to_drop_frames(struct rtmp_stream *stream, bool pframes)
  1153. {
  1154. struct encoder_packet first;
  1155. int64_t buffer_duration_usec;
  1156. size_t num_packets = num_buffered_packets(stream);
  1157. const char *name = pframes ? "p-frames" : "b-frames";
  1158. int priority = pframes ? OBS_NAL_PRIORITY_HIGHEST
  1159. : OBS_NAL_PRIORITY_HIGH;
  1160. int64_t drop_threshold = pframes ? stream->pframe_drop_threshold_usec
  1161. : stream->drop_threshold_usec;
  1162. if (!pframes && stream->dbr_enabled) {
  1163. if (stream->dbr_inc_timeout) {
  1164. uint64_t t = os_gettime_ns();
  1165. if (t >= stream->dbr_inc_timeout) {
  1166. stream->dbr_inc_timeout = 0;
  1167. dbr_inc_bitrate(stream);
  1168. dbr_set_bitrate(stream);
  1169. }
  1170. }
  1171. }
  1172. if (num_packets < 5) {
  1173. if (!pframes)
  1174. stream->congestion = 0.0f;
  1175. return;
  1176. }
  1177. if (!find_first_video_packet(stream, &first))
  1178. return;
  1179. /* if the amount of time stored in the buffered packets waiting to be
  1180. * sent is higher than threshold, drop frames */
  1181. buffer_duration_usec = stream->last_dts_usec - first.dts_usec;
  1182. if (!pframes) {
  1183. stream->congestion =
  1184. (float)buffer_duration_usec / (float)drop_threshold;
  1185. }
  1186. /* alternatively, drop only pframes:
  1187. * (!pframes && stream->dbr_enabled)
  1188. * but let's test without dropping frames
  1189. * at all first */
  1190. if (stream->dbr_enabled) {
  1191. bool bitrate_changed = false;
  1192. if (pframes) {
  1193. return;
  1194. }
  1195. if ((uint64_t)buffer_duration_usec >= DBR_TRIGGER_USEC) {
  1196. pthread_mutex_lock(&stream->dbr_mutex);
  1197. bitrate_changed = dbr_bitrate_lowered(stream);
  1198. pthread_mutex_unlock(&stream->dbr_mutex);
  1199. }
  1200. if (bitrate_changed) {
  1201. debug("buffer_duration_msec: %" PRId64,
  1202. buffer_duration_usec / 1000);
  1203. dbr_set_bitrate(stream);
  1204. }
  1205. return;
  1206. }
  1207. if (buffer_duration_usec > drop_threshold) {
  1208. debug("buffer_duration_usec: %" PRId64, buffer_duration_usec);
  1209. drop_frames(stream, name, priority, pframes);
  1210. }
  1211. }
  1212. static bool add_video_packet(struct rtmp_stream *stream,
  1213. struct encoder_packet *packet)
  1214. {
  1215. if (!silently_reconnecting(stream)) {
  1216. check_to_drop_frames(stream, false);
  1217. check_to_drop_frames(stream, true);
  1218. }
  1219. /* if currently dropping frames, drop packets until it reaches the
  1220. * desired priority */
  1221. if (packet->drop_priority < stream->min_priority) {
  1222. stream->dropped_frames++;
  1223. return false;
  1224. } else {
  1225. stream->min_priority = 0;
  1226. }
  1227. stream->last_dts_usec = packet->dts_usec;
  1228. return add_packet(stream, packet);
  1229. }
  1230. static void rtmp_stream_data(void *data, struct encoder_packet *packet)
  1231. {
  1232. struct rtmp_stream *stream = data;
  1233. struct encoder_packet new_packet;
  1234. bool added_packet = false;
  1235. if (disconnected(stream) || !active(stream))
  1236. return;
  1237. /* encoder fail */
  1238. if (!packet) {
  1239. os_atomic_set_bool(&stream->encode_error, true);
  1240. os_sem_post(stream->send_sem);
  1241. return;
  1242. }
  1243. if (packet->type == OBS_ENCODER_VIDEO) {
  1244. if (!stream->got_first_video) {
  1245. stream->start_dts_offset =
  1246. get_ms_time(packet, packet->dts);
  1247. stream->got_first_video = true;
  1248. }
  1249. obs_parse_avc_packet(&new_packet, packet);
  1250. } else {
  1251. obs_encoder_packet_ref(&new_packet, packet);
  1252. }
  1253. pthread_mutex_lock(&stream->packets_mutex);
  1254. if (!disconnected(stream)) {
  1255. added_packet = (packet->type == OBS_ENCODER_VIDEO)
  1256. ? add_video_packet(stream, &new_packet)
  1257. : add_packet(stream, &new_packet);
  1258. }
  1259. pthread_mutex_unlock(&stream->packets_mutex);
  1260. if (added_packet)
  1261. os_sem_post(stream->send_sem);
  1262. else
  1263. obs_encoder_packet_release(&new_packet);
  1264. }
  1265. static void rtmp_stream_defaults(obs_data_t *defaults)
  1266. {
  1267. obs_data_set_default_int(defaults, OPT_DROP_THRESHOLD, 700);
  1268. obs_data_set_default_int(defaults, OPT_PFRAME_DROP_THRESHOLD, 900);
  1269. obs_data_set_default_int(defaults, OPT_MAX_SHUTDOWN_TIME_SEC, 30);
  1270. obs_data_set_default_string(defaults, OPT_BIND_IP, "default");
  1271. obs_data_set_default_bool(defaults, OPT_NEWSOCKETLOOP_ENABLED, false);
  1272. obs_data_set_default_bool(defaults, OPT_LOWLATENCY_ENABLED, false);
  1273. }
  1274. static obs_properties_t *rtmp_stream_properties(void *unused)
  1275. {
  1276. UNUSED_PARAMETER(unused);
  1277. obs_properties_t *props = obs_properties_create();
  1278. struct netif_saddr_data addrs = {0};
  1279. obs_property_t *p;
  1280. obs_properties_add_int(props, OPT_DROP_THRESHOLD,
  1281. obs_module_text("RTMPStream.DropThreshold"), 200,
  1282. 10000, 100);
  1283. p = obs_properties_add_list(props, OPT_BIND_IP,
  1284. obs_module_text("RTMPStream.BindIP"),
  1285. OBS_COMBO_TYPE_LIST,
  1286. OBS_COMBO_FORMAT_STRING);
  1287. obs_property_list_add_string(p, obs_module_text("Default"), "default");
  1288. netif_get_addrs(&addrs);
  1289. for (size_t i = 0; i < addrs.addrs.num; i++) {
  1290. struct netif_saddr_item item = addrs.addrs.array[i];
  1291. obs_property_list_add_string(p, item.name, item.addr);
  1292. }
  1293. netif_saddr_data_free(&addrs);
  1294. obs_properties_add_bool(props, OPT_NEWSOCKETLOOP_ENABLED,
  1295. obs_module_text("RTMPStream.NewSocketLoop"));
  1296. obs_properties_add_bool(props, OPT_LOWLATENCY_ENABLED,
  1297. obs_module_text("RTMPStream.LowLatencyMode"));
  1298. return props;
  1299. }
  1300. static uint64_t rtmp_stream_total_bytes_sent(void *data)
  1301. {
  1302. struct rtmp_stream *stream = data;
  1303. return stream->total_bytes_sent;
  1304. }
  1305. static int rtmp_stream_dropped_frames(void *data)
  1306. {
  1307. struct rtmp_stream *stream = data;
  1308. return stream->dropped_frames;
  1309. }
  1310. static float rtmp_stream_congestion(void *data)
  1311. {
  1312. struct rtmp_stream *stream = data;
  1313. if (stream->new_socket_loop)
  1314. return (float)stream->write_buf_len /
  1315. (float)stream->write_buf_size;
  1316. else
  1317. return stream->min_priority > 0 ? 1.0f : stream->congestion;
  1318. }
  1319. static int rtmp_stream_connect_time(void *data)
  1320. {
  1321. struct rtmp_stream *stream = data;
  1322. return stream->rtmp.connect_time_ms;
  1323. }
  1324. struct obs_output_info rtmp_output_info = {
  1325. .id = "rtmp_output",
  1326. .flags = OBS_OUTPUT_AV | OBS_OUTPUT_ENCODED | OBS_OUTPUT_SERVICE |
  1327. OBS_OUTPUT_MULTI_TRACK,
  1328. .encoded_video_codecs = "h264",
  1329. .encoded_audio_codecs = "aac",
  1330. .get_name = rtmp_stream_getname,
  1331. .create = rtmp_stream_create,
  1332. .destroy = rtmp_stream_destroy,
  1333. .start = rtmp_stream_start,
  1334. .stop = rtmp_stream_stop,
  1335. .encoded_packet = rtmp_stream_data,
  1336. .get_defaults = rtmp_stream_defaults,
  1337. .get_properties = rtmp_stream_properties,
  1338. .get_total_bytes = rtmp_stream_total_bytes_sent,
  1339. .get_congestion = rtmp_stream_congestion,
  1340. .get_connect_time_ms = rtmp_stream_connect_time,
  1341. .get_dropped_frames = rtmp_stream_dropped_frames,
  1342. };