FS-7500: block overlapping writes

diff --git a/src/switch_core_media.c b/src/switch_core_media.c
index c1f8e7e9e076ec4e7724e2cf1bde823928d4f784..32f40ff384d707b292e6f03945f0a30a0f7f68d2 100644 (file)
--- a/src/switch_core_media.c
+++ b/src/switch_core_media.c
@@ -147,7 +147,7 @@ typedef struct switch_rtp_engine_s {
 
        struct media_helper mh;
        switch_thread_t *media_thread;
-       switch_mutex_t *read_mutex[2];
+
 
        uint8_t reset_codec;
        uint8_t codec_negotiated;
@@ -167,7 +167,8 @@ struct switch_media_handle_s {
        switch_core_media_flag_t media_flags[SCMF_MAX];
        smh_flag_t flags;
        switch_rtp_engine_t engines[SWITCH_MEDIA_TYPE_TOTAL];
-
+       switch_mutex_t *read_mutex[2];
+       switch_mutex_t *write_mutex[2];
        char *codec_order[SWITCH_MAX_CODECS];
     int codec_order_last;
     const switch_codec_implementation_t *codecs[SWITCH_MAX_CODECS];
@@ -1893,14 +1894,14 @@ SWITCH_DECLARE(switch_status_t) switch_core_media_read_lock_unlock(switch_core_s
        }
 
        if (lock) {
-               if (engine->read_mutex[type] && switch_mutex_trylock(engine->read_mutex[type]) != SWITCH_STATUS_SUCCESS) {
+               if (smh->read_mutex[type] && switch_mutex_trylock(smh->read_mutex[type]) != SWITCH_STATUS_SUCCESS) {
                        /* return CNG, another thread is already reading  */
                        switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG1, "%s is already being read for %s\n", 
                                                          switch_channel_get_name(session->channel), type2str(type));
                        return SWITCH_STATUS_INUSE;
                }
        } else {
-               switch_mutex_unlock(engine->read_mutex[type]);
+               switch_mutex_unlock(smh->read_mutex[type]);
        }
 
        return SWITCH_STATUS_SUCCESS;
@@ -1940,7 +1941,7 @@ SWITCH_DECLARE(switch_status_t) switch_core_media_read_frame(switch_core_session
                return SWITCH_STATUS_FALSE;
        }
 
-       if (engine->read_mutex[type] && switch_mutex_trylock(engine->read_mutex[type]) != SWITCH_STATUS_SUCCESS) {
+       if (smh->read_mutex[type] && switch_mutex_trylock(smh->read_mutex[type]) != SWITCH_STATUS_SUCCESS) {
                /* return CNG, another thread is already reading  */
                switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG1, "%s is already being read for %s\n", 
                                                  switch_channel_get_name(session->channel), type2str(type));
@@ -2288,8 +2289,8 @@ SWITCH_DECLARE(switch_status_t) switch_core_media_read_frame(switch_core_session
 
  end:
 
-       if (engine->read_mutex[type]) {
-               switch_mutex_unlock(engine->read_mutex[type]);
+       if (smh->read_mutex[type]) {
+               switch_mutex_unlock(smh->read_mutex[type]);
        }
 
        return status;
@@ -4812,7 +4813,8 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_start_video_thread(switch_co
        switch_thread_cond_create(&v_engine->mh.cond, pool);
        switch_mutex_init(&v_engine->mh.cond_mutex, SWITCH_MUTEX_NESTED, pool);
        switch_mutex_init(&v_engine->mh.file_mutex, SWITCH_MUTEX_NESTED, pool);
-       switch_mutex_init(&v_engine->read_mutex[SWITCH_MEDIA_TYPE_VIDEO], SWITCH_MUTEX_NESTED, pool);
+       switch_mutex_init(&smh->read_mutex[SWITCH_MEDIA_TYPE_VIDEO], SWITCH_MUTEX_NESTED, pool);
+       switch_mutex_init(&smh->write_mutex[SWITCH_MEDIA_TYPE_VIDEO], SWITCH_MUTEX_NESTED, pool);
        switch_thread_create(&v_engine->media_thread, thd_attr, video_helper_thread, &v_engine->mh, switch_core_session_get_pool(session));
 
        return SWITCH_STATUS_SUCCESS;
@@ -5695,7 +5697,8 @@ SWITCH_DECLARE(switch_status_t) switch_core_media_activate_rtp(switch_core_sessi
                uint8_t inb = switch_channel_direction(session->channel) == SWITCH_CALL_DIRECTION_INBOUND;
                const char *ssrc;
 
-               switch_mutex_init(&a_engine->read_mutex[SWITCH_MEDIA_TYPE_AUDIO], SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
+               switch_mutex_init(&smh->read_mutex[SWITCH_MEDIA_TYPE_AUDIO], SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
+               switch_mutex_init(&smh->write_mutex[SWITCH_MEDIA_TYPE_AUDIO], SWITCH_MUTEX_NESTED, switch_core_session_get_pool(session));
 
                //switch_core_media_set_rtp_session(session, SWITCH_MEDIA_TYPE_AUDIO, a_engine->rtp_session);
 
@@ -10017,6 +10020,7 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_write_video_frame(switch_cor
        switch_image_t *img = frame->img;
        switch_status_t encode_status;
        switch_frame_t write_frame = {0};
+       //switch_rtp_engine_t *v_engine;
 
        switch_assert(session);
 
@@ -10041,6 +10045,14 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_write_video_frame(switch_cor
                return SWITCH_STATUS_SUCCESS;
        }
 
+       //v_engine = &smh->engines[SWITCH_MEDIA_TYPE_VIDEO];    
+       if (smh->write_mutex[SWITCH_MEDIA_TYPE_VIDEO] && switch_mutex_lock(smh->write_mutex[SWITCH_MEDIA_TYPE_VIDEO]) != SWITCH_STATUS_SUCCESS) {
+               /* return CNG, another thread is already writing  */
+               switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG1, "%s is already being written to for %s\n", 
+                                                 switch_channel_get_name(session->channel), type2str(SWITCH_MEDIA_TYPE_VIDEO));
+               return SWITCH_STATUS_INUSE;
+       }
+
        if (!smh->video_init && smh->mparams->video_key_first && (now - smh->video_last_key_time) > smh->mparams->video_key_first) {
                switch_core_media_gen_key_frame(session);
 
@@ -10057,7 +10069,8 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_write_video_frame(switch_cor
        }       
 
        if (!img) {
-               return switch_core_session_write_encoded_video_frame(session, frame, flags, stream_id);
+               switch_status_t vstatus = switch_core_session_write_encoded_video_frame(session, frame, flags, stream_id);
+               switch_goto_status(vstatus, done);
        }
 
        write_frame = *frame;
@@ -10100,6 +10113,11 @@ SWITCH_DECLARE(switch_status_t) switch_core_session_write_video_frame(switch_cor
                
        } while(status == SWITCH_STATUS_SUCCESS && encode_status == SWITCH_STATUS_MORE_DATA);
 
+ done:
+
+       if (smh->write_mutex[SWITCH_MEDIA_TYPE_VIDEO]) {
+               switch_mutex_unlock(smh->write_mutex[SWITCH_MEDIA_TYPE_VIDEO]);
+       }
 
        return status;
 }