Changeset 615

Timestamp:

Jul 18, 2006 9:12:24 PM (18 years ago)

Author:

bennylp

Message:

Undo modification to use timer, and back to using threads as the performance is not good

File:

• pjproject/trunk/pjsip-apps/src/samples/siprtp.c (modified) (17 diffs)

pjproject/trunk/pjsip-apps/src/samples/siprtp.c

@@ -123 +123 @@
     pjmedia_rtcp_session rtcp;              /* incoming RTCP session.   */
 
-    /* Timer to send RTP and RTCP: */
-    pj_timer_entry       rtp_timer;         /* timer to send RTP pkt.   */
-    pj_timer_entry       rtcp_timer;        /* timer to send RTCP pkt.  */
+    /* Thread: */
+    pj_bool_t            thread_quit_flag;  /* Stop media thread.       */
+    pj_thread_t         *thread;            /* Media thread.            */
 };
 
@@ -223 +223 @@
 static void on_rx_rtcp(void *user_data, const void *pkt, pj_ssize_t size);
 
-/* This callback is called when it's time to send RTP packet */
-static void on_tx_rtp( pj_timer_heap_t *timer_heap,
-                       struct pj_timer_entry *entry);
-
-/* This callback is called when it's time to send RTCP packet. */
-static void on_tx_rtcp(pj_timer_heap_t *timer_heap,
-                       struct pj_timer_entry *entry);
-
-
 /* Display error */
 static void app_perror(const char *sender, const char *title, 
@@ -397 +388 @@
      * initialized.
      */
-    status = pjmedia_endpt_create(&app.cp.factory, 
-                                  pjsip_endpt_get_ioqueue(app.sip_endpt), 0, 
-                                  &app.med_endpt);
+    status = pjmedia_endpt_create(&app.cp.factory, NULL, 1, &app.med_endpt);
     PJ_ASSERT_RETURN(status == PJ_SUCCESS, status);
 
@@ -419 +408 @@
              * to current port number.
              */
-            struct media_stream *m = &app.call[i].media[j];
             int retry;
 
-            m->call_index = i;
-            m->media_index = j;
-
-            m->rtp_timer.user_data = m;
-            m->rtp_timer.cb = &on_tx_rtp;
-
-            m->rtcp_timer.user_data = m;
-            m->rtcp_timer.cb = &on_tx_rtcp;
-
+            app.call[i].media[j].call_index = i;
+            app.call[i].media[j].media_index = j;
 
             status = -1;
@@ -588 +569 @@
     }
 
+    call = &app.call[i];
+
     /* Verify that we can handle the request. */
     options = 0;
     status = pjsip_inv_verify_request(rdata, &options, NULL, NULL,
-                                      app.sip_endpt, &tdata);
+                                   app.sip_endpt, &tdata);
     if (status != PJ_SUCCESS) {
-
         /*
          * No we can't handle the incoming INVITE request.
          */
-
         if (tdata) {
             pjsip_response_addr res_addr;
-
+            
             pjsip_get_response_addr(tdata->pool, rdata, &res_addr);
-            pjsip_endpt_send_response(app.sip_endpt, &res_addr, tdata, 
-                                      NULL, NULL);
-
+            pjsip_endpt_send_response(app.sip_endpt, &res_addr, tdata,
+                NULL, NULL);
+            
         } else {
-
+            
             /* Respond with 500 (Internal Server Error) */
             pjsip_endpt_respond_stateless(app.sip_endpt, rdata, 500, NULL,
-                                          NULL, NULL);
-        }
-
+                NULL, NULL);
+        }
+        
         return;
-    } 
-
-    call = &app.call[i];
+    }
 
     /* Create UAS dialog */
@@ -803 +782 @@
 
 
-#if defined(PJ_WIN32) && PJ_WIN32 != 0
-#include <windows.h>
-static void boost_priority(void)
-{
-    SetPriorityClass( GetCurrentProcess(), REALTIME_PRIORITY_CLASS);
-    SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_HIGHEST);
-}
-
-#else
-#  define boost_priority()
-#endif
-
-
 /* Worker thread for SIP */
 static int sip_worker_thread(void *arg)
@@ -821 +787 @@
     PJ_UNUSED_ARG(arg);
 
-    boost_priority();
-
     while (!app.thread_quit) {
-        pj_time_val timeout = {0, 1};
+        pj_time_val timeout = {0, 10};
         pjsip_endpt_handle_events(app.sip_endpt, &timeout);
     }
@@ -1083 +1047 @@
 
 
+#if defined(PJ_WIN32) && PJ_WIN32 != 0
+#include <windows.h>
+static void boost_priority(void)
+{
+    SetPriorityClass( GetCurrentProcess(), REALTIME_PRIORITY_CLASS);
+    SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_HIGHEST);
+}
+
+#else
+#  define boost_priority()
+#endif
+
+
 /*
  * This callback is called by media transport on receipt of RTP packet.
@@ -1126 +1103 @@
 }
 
-/* This callback is called when it's time to send RTP packet */
-static void on_tx_rtp( pj_timer_heap_t *timer_heap,
-                       struct pj_timer_entry *entry)
-{
-    pj_status_t status;
-    const pjmedia_rtp_hdr *hdr;
-    pj_ssize_t size;
-    int hdrlen;
-    pj_time_val interval;
-    char packet[512];
-    struct media_stream *strm = entry->user_data;
-
-    PJ_UNUSED_ARG(timer_heap);
-
-    if (!strm->active)
-        return;
-
-    /* Format RTP header */
-    status = pjmedia_rtp_encode_rtp( &strm->out_sess, strm->si.tx_pt,
-                                     0, /* marker bit */
-                                     strm->bytes_per_frame, 
-                                     strm->samples_per_frame,
-                                     (const void**)&hdr, &hdrlen);
-    if (status == PJ_SUCCESS) {
-
-        //PJ_LOG(4,(THIS_FILE, "\t\tTx seq=%d", pj_ntohs(hdr->seq)));
-
-        /* Copy RTP header to packet */
-        pj_memcpy(packet, hdr, hdrlen);
-
-        /* Zero the payload */
-        pj_bzero(packet+hdrlen, strm->bytes_per_frame);
-
-        /* Send RTP packet */
-        size = hdrlen + strm->bytes_per_frame;
-        status = pjmedia_transport_send_rtp(strm->transport, 
-                                            packet, size);
-        if (status != PJ_SUCCESS)
-            app_perror(THIS_FILE, "Error sending RTP packet", status);
-
-    } else {
-        pj_assert(!"RTP encode() error");
-    }
-
-    /* Update RTCP SR */
-    pjmedia_rtcp_tx_rtp( &strm->rtcp, (pj_uint16_t)strm->bytes_per_frame);
-
-    /* Schedule next send */
-    interval.sec = 0;
-    interval.msec = strm->samples_per_frame * 1000 / strm->clock_rate;
-    pj_time_val_normalize(&interval);
-
-    pjsip_endpt_schedule_timer(app.sip_endpt, &strm->rtp_timer, &interval);
-}
-
-
 /*
  * This callback is called by media transport on receipt of RTCP packet.
@@ -1206 +1127 @@
 
 
-/* This callback is called when it's time to send RTCP packet. */
-static void on_tx_rtcp(pj_timer_heap_t *timer_heap,
-                       struct pj_timer_entry *entry)
-{
-    pjmedia_rtcp_pkt *rtcp_pkt;
-    int rtcp_len;
-    pj_ssize_t size;
-    pj_status_t status;
-    pj_time_val interval;
-    struct media_stream *strm = entry->user_data;
-
-    PJ_UNUSED_ARG(timer_heap);
-
-    if (!strm->active)
-        return;
-
-    /* Build RTCP packet */
-    pjmedia_rtcp_build_rtcp(&strm->rtcp, &rtcp_pkt, &rtcp_len);
-
-    /* Send packet */
-    size = rtcp_len;
-    status = pjmedia_transport_send_rtcp(strm->transport,
-                                         rtcp_pkt, size);
-    if (status != PJ_SUCCESS) {
-        app_perror(THIS_FILE, "Error sending RTCP packet", status);
-    }
+/* 
+ * Media thread 
+ *
+ * This is the thread to send and receive both RTP and RTCP packets.
+ */
+static int media_thread(void *arg)
+{
+    enum { RTCP_INTERVAL = 5000, RTCP_RAND = 2000 };
+    struct media_stream *strm = arg;
+    char packet[1500];
+    unsigned msec_interval;
+    pj_timestamp freq, next_rtp, next_rtcp;
+
+
+    /* Boost thread priority if necessary */
+    boost_priority();
+
+    /* Let things settle */
+    pj_thread_sleep(1000);
+
+    msec_interval = strm->samples_per_frame * 1000 / strm->clock_rate;
+    pj_get_timestamp_freq(&freq);
+
+    pj_get_timestamp(&next_rtp);
+    next_rtp.u64 += (freq.u64 * msec_interval / 1000);
+
+    next_rtcp = next_rtp;
+    next_rtcp.u64 += (freq.u64 * (RTCP_INTERVAL+(pj_rand()%RTCP_RAND)) / 1000);
+
+
+    while (!strm->thread_quit_flag) {
+        pj_timestamp now, lesser;
+        pj_time_val timeout;
+        pj_bool_t send_rtp, send_rtcp;
+
+        send_rtp = send_rtcp = PJ_FALSE;
+
+        /* Determine how long to sleep */
+        if (next_rtp.u64 < next_rtcp.u64) {
+            lesser = next_rtp;
+            send_rtp = PJ_TRUE;
+        } else {
+            lesser = next_rtcp;
+            send_rtcp = PJ_TRUE;
+        }
+
+        pj_get_timestamp(&now);
+        if (lesser.u64 <= now.u64) {
+            timeout.sec = timeout.msec = 0;
+            //printf("immediate "); fflush(stdout);
+        } else {
+            pj_uint64_t tick_delay;
+            tick_delay = lesser.u64 - now.u64;
+            timeout.sec = 0;
+            timeout.msec = (pj_uint32_t)(tick_delay * 1000 / freq.u64);
+            pj_time_val_normalize(&timeout);
+
+            //printf("%d:%03d ", timeout.sec, timeout.msec); fflush(stdout);
+        }
+
+        /* Wait for next interval */
+        //if (timeout.sec!=0 && timeout.msec!=0) {
+            pj_thread_sleep(PJ_TIME_VAL_MSEC(timeout));
+            if (strm->thread_quit_flag)
+                break;
+        //}
+
+        pj_get_timestamp(&now);
+
+        if (send_rtp || next_rtp.u64 <= now.u64) {
+            /*
+             * Time to send RTP packet.
+             */
+            pj_status_t status;
+            const pjmedia_rtp_hdr *hdr;
+            pj_ssize_t size;
+            int hdrlen;
+
+            /* Format RTP header */
+            status = pjmedia_rtp_encode_rtp( &strm->out_sess, strm->si.tx_pt,
+                                             0, /* marker bit */
+                                             strm->bytes_per_frame, 
+                                             strm->samples_per_frame,
+                                             (const void**)&hdr, &hdrlen);
+            if (status == PJ_SUCCESS) {
+
+                //PJ_LOG(4,(THIS_FILE, "\t\tTx seq=%d", pj_ntohs(hdr->seq)));
+
+                /* Copy RTP header to packet */
+                pj_memcpy(packet, hdr, hdrlen);
+
+                /* Zero the payload */
+                pj_bzero(packet+hdrlen, strm->bytes_per_frame);
+
+                /* Send RTP packet */
+                size = hdrlen + strm->bytes_per_frame;
+                status = pjmedia_transport_send_rtp(strm->transport, 
+                                                    packet, size);
+                if (status != PJ_SUCCESS)
+                    app_perror(THIS_FILE, "Error sending RTP packet", status);
+
+            } else {
+                pj_assert(!"RTP encode() error");
+            }
+
+            /* Update RTCP SR */
+            pjmedia_rtcp_tx_rtp( &strm->rtcp, (pj_uint16_t)strm->bytes_per_frame);
+
+            /* Schedule next send */
+            next_rtp.u64 += (msec_interval * freq.u64 / 1000);
+        }
+
+
+        if (send_rtcp || next_rtcp.u64 <= now.u64) {
+            /*
+             * Time to send RTCP packet.
+             */
+            pjmedia_rtcp_pkt *rtcp_pkt;
+            int rtcp_len;
+            pj_ssize_t size;
+            pj_status_t status;
+
+            /* Build RTCP packet */
+            pjmedia_rtcp_build_rtcp(&strm->rtcp, &rtcp_pkt, &rtcp_len);
+
     
-    /* Schedule next send */
-    interval.sec = 5;
-    interval.msec = (pj_rand() % 500);
-    pjsip_endpt_schedule_timer(app.sip_endpt, &strm->rtcp_timer, &interval);
-
-}
+            /* Send packet */
+            size = rtcp_len;
+            status = pjmedia_transport_send_rtcp(strm->transport,
+                                                 rtcp_pkt, size);
+            if (status != PJ_SUCCESS) {
+                app_perror(THIS_FILE, "Error sending RTCP packet", status);
+            }
+            
+            /* Schedule next send */
+            next_rtcp.u64 += (freq.u64 * (RTCP_INTERVAL+(pj_rand()%RTCP_RAND)) /
+                              1000);
+        }
+    }
+
+    return 0;
+}
+
 
 /* Callback to be called when SDP negotiation is done in the call: */
@@ -1249 +1280 @@
     const pjmedia_sdp_session *local_sdp, *remote_sdp;
     struct codec *codec_desc = NULL;
-    pj_time_val interval;
     unsigned i;
 
@@ -1257 +1287 @@
 
     /* If this is a mid-call media update, then destroy existing media */
-    if (audio->active)
+    if (audio->thread != NULL)
         destroy_call_media(call->index);
 
@@ -1312 +1342 @@
     status = pjmedia_transport_attach(audio->transport, audio, 
                                       &audio->si.rem_addr, 
-                                      &audio->si.rem_rtcp,
+                                      &audio->si.rem_rtcp, 
                                       sizeof(pj_sockaddr_in),
                                       &on_rx_rtp,
@@ -1321 +1351 @@
     }
 
+    /* Start media thread. */
+    audio->thread_quit_flag = 0;
+    status = pj_thread_create( inv->pool, "media", &media_thread, audio,
+                               0, 0, &audio->thread);
+    if (status != PJ_SUCCESS) {
+        app_perror(THIS_FILE, "Error creating media thread", status);
+        return;
+    }
+
     /* Set the media as active */
     audio->active = PJ_TRUE;
-
-    /* Immediately schedule to send the first RTP packet. */
-    audio->rtp_timer.id = 1;
-    interval.sec = interval.msec = 0;
-    pjsip_endpt_schedule_timer(app.sip_endpt, &audio->rtp_timer, &interval);
-
-    /* And schedule the first RTCP packet */
-    audio->rtcp_timer.id = 1;
-    interval.sec = 4;
-    interval.msec = (pj_rand() % 1000);
-    pjsip_endpt_schedule_timer(app.sip_endpt, &audio->rtcp_timer, &interval);
 }
 
@@ -1343 +1371 @@
     struct media_stream *audio = &app.call[call_index].media[0];
 
-    if (audio->active) {
+    if (audio->thread) {
 
         audio->active = PJ_FALSE;
 
-        if (audio->rtp_timer.id) {
-            audio->rtp_timer.id = 0;
-            pjsip_endpt_cancel_timer(app.sip_endpt, &audio->rtp_timer);
-        }
-
-        if (audio->rtcp_timer.id) {
-            audio->rtcp_timer.id = 0;
-            pjsip_endpt_cancel_timer(app.sip_endpt, &audio->rtcp_timer);
-        }
+        audio->thread_quit_flag = 1;
+        pj_thread_join(audio->thread);
+        pj_thread_destroy(audio->thread);
+        audio->thread = NULL;
+        audio->thread_quit_flag = 0;
 
         pjmedia_transport_detach(audio->transport, audio);
@@ -1361 +1385 @@
 }
 
-
+ 
 /*****************************************************************************
  * USER INTERFACE STUFFS
@@ -1697 +1721 @@
     
     /* Shutting down... */
+    destroy_sip();
     destroy_media();
-    destroy_sip();
 
     if (app.pool) {