Blame view

trunk/src/app/srs_app_recv_thread.cpp 12.0 KB
1 2 3
/*
The MIT License (MIT)
4
Copyright (c) 2013-2015 winlin
5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

Permission is hereby granted, free of charge, to any person obtaining a copy of
this software and associated documentation files (the "Software"), to deal in
the Software without restriction, including without limitation the rights to
use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
the Software, and to permit persons to whom the Software is furnished to do so,
subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/

#include <srs_app_recv_thread.hpp>
26 27
#include <srs_rtmp_sdk.hpp>
#include <srs_rtmp_stack.hpp>
28
#include <srs_app_rtmp_conn.hpp>
29
#include <srs_rtmp_buffer.hpp>
30
#include <srs_kernel_utility.hpp>
31
#include <srs_core_performance.hpp>
32
#include <srs_app_config.hpp>
33
#include <srs_app_source.hpp>
34 35

using namespace std;
36
37 38 39
// the max small bytes to group
#define SRS_MR_SMALL_BYTES 4096
40
ISrsMessageHandler::ISrsMessageHandler()
41 42 43
{
}
44
ISrsMessageHandler::~ISrsMessageHandler()
45 46 47
{
}
48
SrsRecvThread::SrsRecvThread(ISrsMessageHandler* msg_handler, SrsRtmpServer* rtmp_sdk, int timeout_ms)
49
{
50
    timeout = timeout_ms;
51 52
    handler = msg_handler;
    rtmp = rtmp_sdk;
53
    trd = new SrsThread("recv", this, 0, true);
54 55
}
56
SrsRecvThread::~SrsRecvThread()
57
{
58 59
    // stop recv thread.
    stop();
60
61 62
    // destroy the thread.
    srs_freep(trd);
63 64
}
65
int SrsRecvThread::start()
66 67 68 69
{
    return trd->start();
}
70
void SrsRecvThread::stop()
71 72 73 74
{
    trd->stop();
}
75
int SrsRecvThread::cycle()
76 77
{
    int ret = ERROR_SUCCESS;
78
79 80 81 82
    while (trd->can_loop()) {
        if (!handler->can_handle()) {
            st_usleep(timeout * 1000);
            continue;
83
        }
84
    
85
        SrsCommonMessage* msg = NULL;
86
        
87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106
        // recv and handle message
        ret = rtmp->recv_message(&msg);
        if (ret == ERROR_SUCCESS) {
            ret = handler->handle(msg);
        }
    
        if (ret != ERROR_SUCCESS) {
            if (!srs_is_client_gracefully_close(ret)) {
                srs_error("thread process message failed. ret=%d", ret);
            }
    
            // we use no timeout to recv, should never got any error.
            trd->stop_loop();
            
            // notice the handler got a recv error.
            handler->on_recv_error(ret);
    
            return ret;
        }
        srs_verbose("thread loop recv message. ret=%d", ret);
107
    }
108
109 110 111
    return ret;
}
112 113 114 115 116
void SrsRecvThread::stop_loop()
{
    trd->stop_loop();
}
117
void SrsRecvThread::on_thread_start()
118 119 120 121 122 123 124
{
    // the multiple messages writev improve performance large,
    // but the timeout recv will cause 33% sys call performance,
    // to use isolate thread to recv, can improve about 33% performance.
    // @see https://github.com/winlinvip/simple-rtmp-server/issues/194
    // @see: https://github.com/winlinvip/simple-rtmp-server/issues/217
    rtmp->set_recv_timeout(ST_UTIME_NO_TIMEOUT);
125 126
    
    handler->on_thread_start();
127 128
}
129
void SrsRecvThread::on_thread_stop()
130 131
{
    // reset the timeout to pulse mode.
132
    rtmp->set_recv_timeout(timeout * 1000);
133 134
    
    handler->on_thread_stop();
135 136
}
137
SrsQueueRecvThread::SrsQueueRecvThread(SrsConsumer* consumer, SrsRtmpServer* rtmp_sdk, int timeout_ms)
138
    : trd(this, rtmp_sdk, timeout_ms)
139
{
140
    _consumer = consumer;
141
    rtmp = rtmp_sdk;
142
    recv_error_code = ERROR_SUCCESS;
143 144 145 146
}

SrsQueueRecvThread::~SrsQueueRecvThread()
{
147
    stop();
148 149

    // clear all messages.
150
    std::vector<SrsCommonMessage*>::iterator it;
151
    for (it = queue.begin(); it != queue.end(); ++it) {
152
        SrsCommonMessage* msg = *it;
153 154 155 156 157
        srs_freep(msg);
    }
    queue.clear();
}
158 159 160 161 162 163 164 165 166 167
int SrsQueueRecvThread::start()
{
    return trd.start();
}

void SrsQueueRecvThread::stop()
{
    trd.stop();
}
168 169 170 171 172 173 174 175 176 177
bool SrsQueueRecvThread::empty()
{
    return queue.empty();
}

int SrsQueueRecvThread::size()
{
    return (int)queue.size();
}
178
SrsCommonMessage* SrsQueueRecvThread::pump()
179 180 181
{
    srs_assert(!queue.empty());
    
182
    SrsCommonMessage* msg = *queue.begin();
183 184 185 186 187 188
    
    queue.erase(queue.begin());
    
    return msg;
}
189 190 191 192 193
int SrsQueueRecvThread::error_code()
{
    return recv_error_code;
}
194 195 196 197 198 199
bool SrsQueueRecvThread::can_handle()
{
    // we only recv one message and then process it,
    // for the message may cause the thread to stop,
    // when stop, the thread is freed, so the messages
    // are dropped.
zhengfl authored
200
    return empty();
201 202
}
203
int SrsQueueRecvThread::handle(SrsCommonMessage* msg)
204 205 206 207
{
    // put into queue, the send thread will get and process it,
    // @see SrsRtmpConn::process_play_control_msg
    queue.push_back(msg);
zhengfl authored
208 209
#ifdef SRS_PERF_QUEUE_COND_WAIT
    if (_consumer) {
210
        _consumer->wakeup();
zhengfl authored
211 212
    }
#endif
213 214
    return ERROR_SUCCESS;
}
215
216 217
void SrsQueueRecvThread::on_recv_error(int ret)
{
zhengfl authored
218
    recv_error_code = ret;
219 220
#ifdef SRS_PERF_QUEUE_COND_WAIT
    if (_consumer) {
221
        _consumer->wakeup();
222 223
    }
#endif
224 225
}
226 227 228 229 230 231 232 233 234 235 236 237 238 239
void SrsQueueRecvThread::on_thread_start()
{
    // disable the protocol auto response,
    // for the isolate recv thread should never send any messages.
    rtmp->set_auto_response(false);
}

void SrsQueueRecvThread::on_thread_stop()
{
    // enable the protocol auto response,
    // for the isolate recv thread terminated.
    rtmp->set_auto_response(true);
}
240
SrsPublishRecvThread::SrsPublishRecvThread(
241 242
    SrsRtmpServer* rtmp_sdk, 
    SrsRequest* _req, int mr_sock_fd, int timeout_ms, 
243 244 245
    SrsRtmpConn* conn, SrsSource* source, bool is_fmle, bool is_edge
): trd(this, rtmp_sdk, timeout_ms)
{
246
    rtmp = rtmp_sdk;
247
248 249 250 251 252 253 254
    _conn = conn;
    _source = source;
    _is_fmle = is_fmle;
    _is_edge = is_edge;

    recv_error_code = ERROR_SUCCESS;
    _nb_msgs = 0;
255
    error = st_cond_new();
256 257 258
    
    req = _req;
    mr_fd = mr_sock_fd;
259
260 261 262 263 264
    // the mr settings, 
    // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
    mr = _srs_config->get_mr_enabled(req->vhost);
    mr_sleep = _srs_config->get_mr_sleep_ms(req->vhost);
    
265 266
    realtime = _srs_config->get_realtime_enabled(req->vhost);
    
267
    _srs_config->subscribe(this);
268 269 270 271
}

SrsPublishRecvThread::~SrsPublishRecvThread()
{
272 273
    _srs_config->unsubscribe(this);
    
274
    trd.stop();
275 276 277 278 279 280 281 282 283 284 285 286 287
    st_cond_destroy(error);
}

int SrsPublishRecvThread::wait(int timeout_ms)
{
    if (recv_error_code != ERROR_SUCCESS) {
        return recv_error_code;
    }
    
    // ignore any return of cond wait.
    st_cond_timedwait(error, timeout_ms * 1000);
    
    return ERROR_SUCCESS;
288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309
}

int64_t SrsPublishRecvThread::nb_msgs()
{
    return _nb_msgs;
}

int SrsPublishRecvThread::error_code()
{
    return recv_error_code;
}

int SrsPublishRecvThread::start()
{
    return trd.start();
}

void SrsPublishRecvThread::stop()
{
    trd.stop();
}
310 311 312 313 314
void SrsPublishRecvThread::on_thread_start()
{
    // we donot set the auto response to false,
    // for the main thread never send message.
315
#ifdef SRS_PERF_MERGED_READ
316 317 318 319 320 321 322 323
    if (mr) {
        // set underlayer buffer size
        set_socket_buffer(mr_sleep);

        // disable the merge read
        // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
        rtmp->set_merge_read(true, this);
    }
324
#endif
325 326 327 328 329 330 331 332 333 334 335
}

void SrsPublishRecvThread::on_thread_stop()
{
    // we donot set the auto response to true,
    // for we donot set to false yet.
    
    // when thread stop, signal the conn thread which wait.
    // @see https://github.com/winlinvip/simple-rtmp-server/issues/244
    st_cond_signal(error);
336
#ifdef SRS_PERF_MERGED_READ
337 338 339 340 341
    if (mr) {
        // disable the merge read
        // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
        rtmp->set_merge_read(false, NULL);
    }
342
#endif
343 344
}
345 346 347 348 349 350
bool SrsPublishRecvThread::can_handle()
{
    // publish thread always can handle message.
    return true;
}
351
int SrsPublishRecvThread::handle(SrsCommonMessage* msg)
352 353 354 355
{
    int ret = ERROR_SUCCESS;

    _nb_msgs++;
356 357 358 359
    
    // log to show the time of recv thread.
    srs_verbose("recv thread now=%"PRId64"us, got msg time=%"PRId64"ms, size=%d",
        srs_update_system_time_ms(), msg->header.timestamp, msg->size);
360
361 362
    // the rtmp connection will handle this message
    ret = _conn->handle_publish_message(_source, msg, _is_fmle, _is_edge);
363 364 365 366

    // must always free it,
    // the source will copy it if need to use.
    srs_freep(msg);
367
    
368 369
    return ret;
}
370 371 372 373

void SrsPublishRecvThread::on_recv_error(int ret)
{
    recv_error_code = ret;
374 375 376 377

    // when recv thread error, signal the conn thread to process it.
    // @see https://github.com/winlinvip/simple-rtmp-server/issues/244
    st_cond_signal(error);
378
}
379
380
#ifdef SRS_PERF_MERGED_READ
381
void SrsPublishRecvThread::on_read(ssize_t nread)
382
{
383
    if (!mr || realtime) {
384 385 386 387
        return;
    }
    
    if (nread < 0 || mr_sleep <= 0) {
388 389
        return;
    }
390
    
391 392 393 394 395 396
    /**
    * to improve read performance, merge some packets then read,
    * when it on and read small bytes, we sleep to wait more data.,
    * that is, we merge some data to read together.
    * @see https://github.com/winlinvip/simple-rtmp-server/issues/241
    */
397
    if (nread < SRS_MR_SMALL_BYTES) {
398
        st_usleep(mr_sleep * 1000);
399
    }
400
}
401
#endif
402 403 404 405

int SrsPublishRecvThread::on_reload_vhost_mr(string vhost)
{
    int ret = ERROR_SUCCESS;
406 407 408 409
    
    if (req->vhost != vhost) {
        return ret;
    }
410 411 412 413 414 415

    // the mr settings, 
    // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
    bool mr_enabled = _srs_config->get_mr_enabled(req->vhost);
    int sleep_ms = _srs_config->get_mr_sleep_ms(req->vhost);
416 417 418 419 420
    // update buffer when sleep ms changed.
    if (mr_sleep != sleep_ms) {
        set_socket_buffer(sleep_ms);
    }
421
#ifdef SRS_PERF_MERGED_READ
422 423 424 425 426 427 428 429 430 431 432
    // mr enabled=>disabled
    if (mr && !mr_enabled) {
        // disable the merge read
        // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
        rtmp->set_merge_read(false, NULL);
    }
    // mr disabled=>enabled
    if (!mr && mr_enabled) {
        // enable the merge read
        // @see https://github.com/winlinvip/simple-rtmp-server/issues/241
        rtmp->set_merge_read(true, this);
433 434 435 436 437 438 439
    }
#endif

    // update to new state
    mr = mr_enabled;
    mr_sleep = sleep_ms;
    
440 441 442
    return ret;
}
443 444 445 446 447 448 449 450 451 452 453 454 455 456 457
int SrsPublishRecvThread::on_reload_vhost_realtime(string vhost)
{
    int ret = ERROR_SUCCESS;
    
    if (req->vhost != vhost) {
        return ret;
    }
    
    bool realtime_enabled = _srs_config->get_realtime_enabled(req->vhost);
    srs_trace("realtime changed %d=>%d", realtime, realtime_enabled);
    realtime = realtime_enabled;
    
    return ret;
}
458 459
void SrsPublishRecvThread::set_socket_buffer(int sleep_ms)
{
460
    // the bytes:
461 462
    //      4KB=4096, 8KB=8192, 16KB=16384, 32KB=32768, 64KB=65536,
    //      128KB=131072, 256KB=262144, 512KB=524288
463
    // the buffer should set to sleep*kbps/8,
464 465 466
    // for example, your system delivery stream in 1000kbps,
    // sleep 800ms for small bytes, the buffer should set to:
    //      800*1000/8=100000B(about 128KB).
467
    // other examples:
468
    //      2000*3000/8=750000B(about 732KB).
469 470
    //      2000*5000/8=1250000B(about 1220KB).
    int kbps = 5000;
471
    int socket_buffer_size = sleep_ms * kbps / 8;
472 473 474 475 476
    
    int fd = mr_fd;
    int onb_rbuf = 0;
    socklen_t sock_buf_size = sizeof(int);
    getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &onb_rbuf, &sock_buf_size);
477 478 479

    // socket recv buffer, system will double it.
    int nb_rbuf = socket_buffer_size / 2;
480
    if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &nb_rbuf, sock_buf_size) < 0) {
481
        srs_warn("set sock SO_RCVBUF=%d failed.", nb_rbuf);
482
    }
483
    getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &nb_rbuf, &sock_buf_size);
484
485 486 487
    srs_trace("mr change sleep %d=>%d, erbuf=%d, rbuf %d=>%d, sbytes=%d, realtime=%d",
        mr_sleep, sleep_ms, socket_buffer_size, onb_rbuf, nb_rbuf, 
        SRS_MR_SMALL_BYTES, realtime);
488 489

    rtmp->set_recv_buffer(nb_rbuf);
490 491
}