Threads-Hybrid: Use ST wait and IO for RECV/SEND thread.

winlinvip · winlinvip · commit 5074a4d44473 · 2021-04-26T08:14:31.000+08:00
diff --git a/trunk/src/app/srs_app_listener.cpp b/trunk/src/app/srs_app_listener.cpp
@@ -331,10 +331,8 @@ int SrsUdpMuxSocket::recvfrom(srs_utime_t timeout)
 
 int SrsUdpMuxSocket::raw_recvfrom()
 {
-    int osfd = srs_netfd_fileno(lfd);
-
     fromlen = sizeof(from);
-    nread = ::recvfrom(osfd, buf, nb_buf, 0, (sockaddr*)&from, (socklen_t*)&fromlen);
+    nread = srs_recvfrom(lfd, buf, nb_buf, (sockaddr*)&from, &fromlen, SRS_UTIME_NO_TIMEOUT);
     if (nread <= 0) {
         return nread;
     }
@@ -389,8 +387,8 @@ srs_error_t SrsUdpMuxSocket::sendto(void* data, int size, srs_utime_t timeout)
     if (nb_write <= 0) {
         if (nb_write < 0 && errno == ETIME) {
             return srs_error_new(ERROR_SOCKET_TIMEOUT, "sendto timeout %d ms", srsu2msi(timeout));
-        }   
-    
+        }
+
         return srs_error_new(ERROR_SOCKET_WRITE, "sendto");
     }
 
@@ -406,8 +404,7 @@ srs_error_t SrsUdpMuxSocket::sendto(void* data, int size, srs_utime_t timeout)
 
 int SrsUdpMuxSocket::raw_sendto(void* data, int size)
 {
-    int osfd = srs_netfd_fileno(lfd);
-    return ::sendto(osfd, data, size, 0, (sockaddr*)&from, (socklen_t)fromlen);
+    return srs_sendto(lfd, data, size, (sockaddr*)&from, fromlen, SRS_UTIME_NO_TIMEOUT);
 }
 
 srs_netfd_t SrsUdpMuxSocket::stfd()
diff --git a/trunk/src/app/srs_app_threads.cpp b/trunk/src/app/srs_app_threads.cpp
@@ -1903,9 +1903,6 @@ srs_error_t SrsAsyncRecvManager::do_start()
 {
     srs_error_t err = srs_success;
 
-    // TODO: FIXME: Config it?
-    srs_utime_t interval = 10 * SRS_UTIME_MILLISECONDS;
-
     // We must deep copy listeners if changed.
     vector<SrsThreadUdpListener*> listeners;
 
@@ -1956,11 +1953,8 @@ srs_error_t SrsAsyncRecvManager::do_start()
             continue;
         }
 
-        // TODO: FIXME: Maybe we should use cond wait?
-        timespec tv = {0};
-        tv.tv_sec = interval / SRS_UTIME_SECONDS;
-        tv.tv_nsec = (interval % SRS_UTIME_SECONDS) * 1000;
-        nanosleep(&tv, NULL);
+        // OK to use ST wait.
+        srs_usleep(10 * SRS_UTIME_MILLISECONDS);
     }
 
     return err;
@@ -2100,9 +2094,6 @@ srs_error_t SrsAsyncSendManager::do_start()
 {
     srs_error_t err = srs_success;
 
-    // TODO: FIXME: Config it?
-    srs_utime_t interval = 10 * SRS_UTIME_MILLISECONDS;
-
     while (true) {
         vector<SrsAsyncUdpPacket*> flying_sending_packets;
         sending_packets_->swap(flying_sending_packets);
@@ -2123,11 +2114,8 @@ srs_error_t SrsAsyncSendManager::do_start()
             continue;
         }
 
-        // TODO: FIXME: Maybe we should use cond wait?
-        timespec tv = {0};
-        tv.tv_sec = interval / SRS_UTIME_SECONDS;
-        tv.tv_nsec = (interval % SRS_UTIME_SECONDS) * 1000;
-        nanosleep(&tv, NULL);
+        // OK to use ST wait.
+        srs_usleep(10 * SRS_UTIME_MILLISECONDS);
     }
 
     return err;

Original file line number	Diff line number	Diff line change
`@@ -331,10 +331,8 @@ int SrsUdpMuxSocket::recvfrom(srs_utime_t timeout)`
`331`	`331`
`332`	`332`	`int SrsUdpMuxSocket::raw_recvfrom()`
`333`	`333`	`{`
`334`		`- int osfd = srs_netfd_fileno(lfd);`
`335`		`-`
`336`	`334`	`fromlen = sizeof(from);`
`337`		`- nread = ::recvfrom(osfd, buf, nb_buf, 0, (sockaddr)&from, (socklen_t)&fromlen);`
	`335`	`+ nread = srs_recvfrom(lfd, buf, nb_buf, (sockaddr*)&from, &fromlen, SRS_UTIME_NO_TIMEOUT);`
`338`	`336`	`if (nread <= 0) {`
`339`	`337`	`return nread;`
`340`	`338`	`}`
`@@ -389,8 +387,8 @@ srs_error_t SrsUdpMuxSocket::sendto(void* data, int size, srs_utime_t timeout)`
`389`	`387`	`if (nb_write <= 0) {`
`390`	`388`	`if (nb_write < 0 && errno == ETIME) {`
`391`	`389`	`return srs_error_new(ERROR_SOCKET_TIMEOUT, "sendto timeout %d ms", srsu2msi(timeout));`
`392`		`- }`
`393`		`-`
	`390`	`+ }`
	`391`	`+`
`394`	`392`	`return srs_error_new(ERROR_SOCKET_WRITE, "sendto");`
`395`	`393`	`}`
`396`	`394`
`@@ -406,8 +404,7 @@ srs_error_t SrsUdpMuxSocket::sendto(void* data, int size, srs_utime_t timeout)`
`406`	`404`
`407`	`405`	`int SrsUdpMuxSocket::raw_sendto(void* data, int size)`
`408`	`406`	`{`
`409`		`- int osfd = srs_netfd_fileno(lfd);`
`410`		`- return ::sendto(osfd, data, size, 0, (sockaddr*)&from, (socklen_t)fromlen);`
	`407`	`+ return srs_sendto(lfd, data, size, (sockaddr*)&from, fromlen, SRS_UTIME_NO_TIMEOUT);`
`411`	`408`	`}`
`412`	`409`
`413`	`410`	`srs_netfd_t SrsUdpMuxSocket::stfd()`