Changeset 4678 for pjproject

Timestamp:

Dec 5, 2013 3:03:36 AM (11 years ago)

Author:

bennylp

Message:

Re #1519: fixed threading issues on Python. On Python, only threads created by Python can call Python. This creates problem with calling callback from worker thread. The SIP worker thread can be disabled, but we have other worker threads such as the sound device that cannot be disabled. The solution in this patch is to create small framework to post a job to "main thread" during libHandleEvents(). The main thread is thread that calls libCreate().

Location:

pjproject/branches/projects/pjsua2/pjsip

Files:

• include/pjsua2/endpoint.hpp (modified) (8 diffs)
• src/pjsua2/endpoint.cpp (modified) (14 diffs)
• src/pjsua2/media.cpp (modified) (3 diffs)

pjproject/branches/projects/pjsua2/pjsip/include/pjsua2/endpoint.hpp

@@ -27 +27 @@
 #include <pjsua2/media.hpp>
 #include <pjsua2/siptypes.hpp>
+#include <list>
 
 /** PJSUA2 API is inside pj namespace */
@@ -184 +185 @@
      */
     unsigned            threadCnt;
+
+    /**
+     * When this flag is non-zero, all callbacks that come from thread
+     * other than main thread will be posted to the main thread and
+     * to be executed by Endpoint::libHandleEvents() function. This
+     * includes the logging callback. Note that this will only work if
+     * threadCnt is set to zero and Endpoint::libHandleEvents() is
+     * performed by main thread. By default, the main thread is set
+     * from the thread that invoke Endpoint::libCreate()
+     *
+     * Default: false
+     */
+    bool                mainThreadOnly;
 
     /**
@@ -635 +649 @@
 };
 
+/* This represents posted job */
+struct PendingJob
+{
+    /** Perform the job */
+    virtual void execute(bool is_pending) = 0;
+
+    /** Virtual destructor */
+    virtual ~PendingJob() {}
+};
+
 //////////////////////////////////////////////////////////////////////////////
 
@@ -717 +741 @@
      * structure, because polling then will be done by these worker threads
      * instead.
+     *
+     * If EpConfig::UaConfig::mainThreadOnly is enabled and this function
+     * is called from the main thread (by default the main thread is thread
+     * that calls libCreate()), this function will also scan and run any
+     * pending jobs in the list.
      *
      * @param msec_timeout Maximum time to wait, in miliseconds.
@@ -765 +794 @@
 
     /**
+     * Write a log entry.
+     *
+     * @param e                 The log entry.
+     */
+    void utilLogWrite(LogEntry &e);
+
+    /**
      * This is a utility function to verify that valid SIP url is given. If the
      * URL is a valid SIP/SIPS scheme, PJ_SUCCESS will be returned.
@@ -814 +850 @@
      */
     void utilTimerCancel(Token prmToken);
+
+    /**
+     * Utility to register a pending job to be executed by main thread.
+     * If EpConfig::UaConfig::mainThreadOnly is false, the job will be
+     * executed immediately.
+     *
+     * @param job               The job class.
+     */
+    void utilAddPendingJob(PendingJob *job);
 
     /**
@@ -1122 +1167 @@
     { PJ_UNUSED_ARG(prm); }
 
-
 private:
     static Endpoint             *instance_;     // static instance
@@ -1129 +1173 @@
     AudDevManager                audioDevMgr;
     CodecInfoVector              codecInfoList;
+
+    /* Pending logging */
+    bool                         mainThreadOnly;
+    void                        *mainThread;
+    unsigned                     pendingJobSize;
+    std::list<PendingJob*>       pendingJobs;
+
+    void performPendingJobs();
 
     /* Endpoint static callbacks */

pjproject/branches/projects/pjsua2/pjsip/src/pjsua2/endpoint.cpp

@@ -112 +112 @@
     NODE_READ_UNSIGNED( this_node, maxCalls);
     NODE_READ_UNSIGNED( this_node, threadCnt);
+    NODE_READ_BOOL    ( this_node, mainThreadOnly);
     NODE_READ_STRINGV ( this_node, nameserver);
     NODE_READ_STRING  ( this_node, userAgent);
@@ -126 +127 @@
     NODE_WRITE_UNSIGNED( this_node, maxCalls);
     NODE_WRITE_UNSIGNED( this_node, threadCnt);
+    NODE_WRITE_BOOL    ( this_node, mainThreadOnly);
     NODE_WRITE_STRINGV ( this_node, nameserver);
     NODE_WRITE_STRING  ( this_node, userAgent);
@@ -342 +344 @@
 
 ///////////////////////////////////////////////////////////////////////////////
+/* Class to post log to main thread */
+struct PendingLog : public PendingJob
+{
+    LogEntry entry;
+    virtual void execute(bool is_pending)
+    {
+        PJ_UNUSED_ARG(is_pending);
+        Endpoint::instance().utilLogWrite(entry);
+    }
+};
+
+///////////////////////////////////////////////////////////////////////////////
 /*
  * Endpoint instance
  */
 Endpoint::Endpoint()
-: writer(NULL)
+: writer(NULL), mainThreadOnly(false), mainThread(NULL), pendingJobSize(0)
 {
     if (instance_) {
@@ -365 +379 @@
 Endpoint::~Endpoint()
 {
+    while (!pendingJobs.empty()) {
+        delete pendingJobs.front();
+        pendingJobs.pop_front();
+    }
+
     try {
         libDestroy();
@@ -383 +402 @@
 }
 
+void Endpoint::utilAddPendingJob(PendingJob *job)
+{
+    enum {
+        MAX_PENDING_JOBS = 1024
+    };
+
+    /* See if we can execute immediately */
+    if (!mainThreadOnly || pj_thread_this()==mainThread) {
+        job->execute(false);
+        delete job;
+        return;
+    }
+
+    if (pendingJobSize > MAX_PENDING_JOBS) {
+        enum { NUMBER_TO_DISCARD = 5 };
+
+        pj_enter_critical_section();
+        for (unsigned i=0; i<NUMBER_TO_DISCARD; ++i) {
+            delete pendingJobs.back();
+            pendingJobs.pop_back();
+        }
+
+        pendingJobSize -= NUMBER_TO_DISCARD;
+        pj_leave_critical_section();
+
+        utilLogWrite(1, THIS_FILE,
+                     "*** ERROR: Job queue full!! Jobs discarded!!! ***");
+    }
+
+    pj_enter_critical_section();
+    pendingJobs.push_back(job);
+    pendingJobSize++;
+    pj_leave_critical_section();
+}
+
+/* Handle log callback */
+void Endpoint::utilLogWrite(LogEntry &entry)
+{
+    if (mainThreadOnly && pj_thread_this() != mainThread) {
+        PendingLog *job = new PendingLog;
+        job->entry = entry;
+        utilAddPendingJob(job);
+    } else {
+        writer->write(entry);
+    }
+}
+
+/* Run pending jobs only in main thread */
+void Endpoint::performPendingJobs()
+{
+    if (pj_thread_this() != mainThread)
+        return;
+
+    if (pendingJobSize == 0)
+        return;
+
+    for (;;) {
+        PendingJob *job = NULL;
+
+        pj_enter_critical_section();
+        if (pendingJobSize != 0) {
+            job = pendingJobs.front();
+            pendingJobs.pop_front();
+            pendingJobSize--;
+        }
+        pj_leave_critical_section();
+
+        if (job) {
+            job->execute(true);
+            delete job;
+        } else
+            break;
+    }
+}
+
 ///////////////////////////////////////////////////////////////////////////////
 /*
@@ -400 +494 @@
     entry.threadName = string(pj_thread_get_name(pj_thread_this()));
 
-    ep.writer->write(entry);
+    ep.utilLogWrite(entry);
 }
 
@@ -837 +931 @@
 }
 
+struct PendingOnDtmfDigitCallback : public PendingJob
+{
+    int call_id;
+    OnDtmfDigitParam prm;
+
+    virtual void execute(bool is_pending)
+    {
+        PJ_UNUSED_ARG(is_pending);
+
+        Call *call = Call::lookup(call_id);
+        if (!call)
+            return;
+
+        call->onDtmfDigit(prm);
+    }
+};
+
 void Endpoint::on_dtmf_digit(pjsua_call_id call_id, int digit)
 {
@@ -844 +955 @@
     }
     
-    OnDtmfDigitParam prm;
+    PendingOnDtmfDigitCallback *job = new PendingOnDtmfDigitCallback;
+    job->call_id = call_id;
     char buf[10];
     pj_ansi_sprintf(buf, "%c", digit);
-    prm.digit = (string)buf;
-    
-    call->onDtmfDigit(prm);
+    job->prm.digit = (string)buf;
+    
+    Endpoint::instance().utilAddPendingJob(job);
 }
 
@@ -982 +1094 @@
 }
 
+
+struct PendingOnMediaTransportCallback : public PendingJob
+{
+    int call_id;
+    OnCallMediaTransportStateParam prm;
+
+    virtual void execute(bool is_pending)
+    {
+        PJ_UNUSED_ARG(is_pending);
+
+        Call *call = Call::lookup(call_id);
+        if (!call)
+            return;
+
+        call->onCallMediaTransportState(prm);
+    }
+};
+
 pj_status_t
 Endpoint::on_call_media_transport_state(pjsua_call_id call_id,
@@ -990 +1120 @@
         return PJ_SUCCESS;
     }
-    
-    OnCallMediaTransportStateParam prm;
-    prm.medIdx = info->med_idx;
-    prm.state = info->state;
-    prm.status = info->status;
-    prm.sipErrorCode = info->sip_err_code;
-    
-    call->onCallMediaTransportState(prm);
-    
+
+    PendingOnMediaTransportCallback *job = new PendingOnMediaTransportCallback;
+    
+    job->call_id = call_id;
+    job->prm.medIdx = info->med_idx;
+    job->prm.state = info->state;
+    job->prm.status = info->status;
+    job->prm.sipErrorCode = info->sip_err_code;
+    
+    Endpoint::instance().utilAddPendingJob(job);
+
     return PJ_SUCCESS;
 }
+
+struct PendingOnMediaEventCallback : public PendingJob
+{
+    int call_id;
+    OnCallMediaEventParam prm;
+
+    virtual void execute(bool is_pending)
+    {
+        Call *call = Call::lookup(call_id);
+        if (!call)
+            return;
+
+        if (is_pending) {
+            /* Can't do this anymore, pointer is invalid */
+            prm.ev.pjMediaEvent = NULL;
+        }
+
+        call->onCallMediaEvent(prm);
+    }
+};
 
 void Endpoint::on_call_media_event(pjsua_call_id call_id,
@@ -1011 +1163 @@
     }
     
-    OnCallMediaEventParam prm;
-    prm.medIdx = med_idx;
-    prm.ev.fromPj(*event);
-    
-    call->onCallMediaEvent(prm);
+    PendingOnMediaEventCallback *job = new PendingOnMediaEventCallback;
+
+    job->call_id = call_id;
+    job->prm.medIdx = med_idx;
+    job->prm.ev.fromPj(*event);
+    
+    Endpoint::instance().utilAddPendingJob(job);
 }
 
@@ -1058 +1212 @@
 {
     PJSUA2_CHECK_EXPR( pjsua_create() );
+    mainThread = pj_thread_this();
 }
 
@@ -1080 +1235 @@
         log_cfg.cb = &Endpoint::logFunc;
     }
+    mainThreadOnly = prmEpConfig.uaConfig.mainThreadOnly;
 
     /* Setup UA callbacks */
@@ -1136 +1292 @@
 int Endpoint::libHandleEvents(unsigned msec_timeout)
 {
+    performPendingJobs();
     return pjsua_handle_events(msec_timeout);
 }

pjproject/branches/projects/pjsua2/pjsip/src/pjsua2/media.cpp

@@ -377 +377 @@
 
 AudDevManager::AudDevManager()
-{
-    devMedia = new DevAudioMedia;
+: devMedia(NULL)
+{
 }
 
@@ -394 +394 @@
 AudioMedia &AudDevManager::getCaptureDevMedia() throw(Error)
 {
+    if (!devMedia)
+        devMedia = new DevAudioMedia;
     return *devMedia;
 }
@@ -404 +406 @@
 AudioMedia &AudDevManager::getPlaybackDevMedia() throw(Error)
 {
+    if (!devMedia)
+        devMedia = new DevAudioMedia;
     return *devMedia;
 }