rtmp-stream.c 42 KB

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