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

git-svn-id: http://svn.pjsip.org/repos/pjproject/trunk@615 74dad513-b988-da41-8d7b-12977e46ad98

1 files changed, 185 insertions, 161 deletions

diff --git a/pjsip-apps/src/samples/siprtp.c b/pjsip-apps/src/samples/siprtp.c
index 588350fb..724d1c27 100644
--- a/pjsip-apps/src/samples/siprtp.c
+++ b/pjsip-apps/src/samples/siprtp.c
@@ -122,9 +122,9 @@ struct media_stream
     /* RTCP stats: */
     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.		*/
 };
 
 
@@ -222,15 +222,6 @@ static void on_rx_rtp(void *user_data, const void *pkt, pj_ssize_t size);
 /* This callback is called by media transport on receipt of RTCP packet. */
 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, 
 		       pj_status_t status);
@@ -396,9 +387,7 @@ static pj_status_t init_media()
     /* Initialize media endpoint so that at least error subsystem is properly
      * 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);
 
 
@@ -418,18 +407,10 @@ static pj_status_t init_media()
 	    /* Repeat binding media socket to next port when fails to bind
 	     * 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;
 	    for (retry=0; retry<100; ++retry,rtp_port+=2)  {
@@ -587,34 +568,32 @@ static void process_incoming_call(pjsip_rx_data *rdata)
 	return;
     }
 
+    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 */
     status = pjsip_dlg_create_uas( pjsip_ua_instance(), rdata,
@@ -802,28 +781,13 @@ static void app_perror(const char *sender, const char *title,
 }
 
 
-#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)
 {
     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);
     }
 
@@ -1082,6 +1046,19 @@ static pj_status_t create_sdp( pj_pool_t *pool,
 }
 
 
+#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.
  */
@@ -1125,62 +1102,6 @@ static void on_rx_rtp(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)
-{
-    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.
  */
@@ -1205,40 +1126,150 @@ 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 RTCP packet. */
-static void on_tx_rtcp(pj_timer_heap_t *timer_heap,
-		       struct pj_timer_entry *entry)
+/* 
+ * Media thread 
+ *
+ * This is the thread to send and receive both RTP and RTCP packets.
+ */
+static int media_thread(void *arg)
 {
-    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;
+    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;
 
-    PJ_UNUSED_ARG(timer_heap);
 
-    if (!strm->active)
-	return;
+    /* Boost thread priority if necessary */
+    boost_priority();
 
-    /* Build RTCP packet */
-    pjmedia_rtcp_build_rtcp(&strm->rtcp, &rtcp_pkt, &rtcp_len);
+    /* 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);
 
-    /* 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 */
-    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: */
 static void call_on_media_update( pjsip_inv_session *inv,
 				  pj_status_t status)
@@ -1248,7 +1279,6 @@ static void call_on_media_update( pjsip_inv_session *inv,
     struct media_stream *audio;
     const pjmedia_sdp_session *local_sdp, *remote_sdp;
     struct codec *codec_desc = NULL;
-    pj_time_val interval;
     unsigned i;
 
     call = inv->mod_data[mod_siprtp.id];
@@ -1256,7 +1286,7 @@ static void call_on_media_update( pjsip_inv_session *inv,
     audio = &call->media[0];
 
     /* If this is a mid-call media update, then destroy existing media */
-    if (audio->active)
+    if (audio->thread != NULL)
 	destroy_call_media(call->index);
 
 
@@ -1311,7 +1341,7 @@ static void call_on_media_update( pjsip_inv_session *inv,
     /* Attach media to transport */
     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,
 				      &on_rx_rtcp);
@@ -1320,19 +1350,17 @@ static void call_on_media_update( pjsip_inv_session *inv,
 	return;
     }
 
+    /* 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);
 }
 
 
@@ -1342,25 +1370,21 @@ static void destroy_call_media(unsigned call_index)
 {
     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);
     }
 }
 
-
+ 
 /*****************************************************************************
  * USER INTERFACE STUFFS
  */
@@ -1696,8 +1720,8 @@ int main(int argc, char *argv[])
 
     
     /* Shutting down... */
-    destroy_media();
     destroy_sip();
+    destroy_media();
 
     if (app.pool) {
 	pj_pool_release(app.pool);