socket_base.cpp 28.1 KB
Newer Older
1
/*
Martin Sustrik's avatar
Martin Sustrik committed
2
    Copyright (c) 2009-2011 250bpm s.r.o.
3
    Copyright (c) 2007-2009 iMatix Corporation
4
    Copyright (c) 2011 VMware, Inc.
5
    Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
6 7 8 9

    This file is part of 0MQ.

    0MQ is free software; you can redistribute it and/or modify it under
10
    the terms of the GNU Lesser General Public License as published by
11 12 13 14 15 16
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.

    0MQ is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17
    GNU Lesser General Public License for more details.
18

19
    You should have received a copy of the GNU Lesser General Public License
20 21 22
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

23
#include <new>
24
#include <string>
25 26
#include <algorithm>

27 28 29 30 31
#include "platform.hpp"

#if defined ZMQ_HAVE_WINDOWS
#include "windows.hpp"
#if defined _MSC_VER
boris@boressoft.ru's avatar
boris@boressoft.ru committed
32 33 34
#if defined WINCE
#include <cmnintrin.h>
#else
35 36
#include <intrin.h>
#endif
boris@boressoft.ru's avatar
boris@boressoft.ru committed
37
#endif
38 39 40
#else
#include <unistd.h>
#endif
41

42
#include "socket_base.hpp"
43
#include "tcp_listener.hpp"
44
#include "ipc_listener.hpp"
45
#include "tcp_connecter.hpp"
46
#include "io_thread.hpp"
47
#include "session_base.hpp"
Martin Sustrik's avatar
Martin Sustrik committed
48
#include "config.hpp"
Martin Sustrik's avatar
Martin Sustrik committed
49
#include "pipe.hpp"
50
#include "err.hpp"
51
#include "ctx.hpp"
malosek's avatar
malosek committed
52
#include "platform.hpp"
53
#include "likely.hpp"
54
#include "msg.hpp"
55 56 57
#include "address.hpp"
#include "ipc_address.hpp"
#include "tcp_address.hpp"
58 59 60
#ifdef ZMQ_HAVE_OPENPGM
#include "pgm_socket.hpp"
#endif
61

62 63 64 65 66 67 68
#include "pair.hpp"
#include "pub.hpp"
#include "sub.hpp"
#include "req.hpp"
#include "rep.hpp"
#include "pull.hpp"
#include "push.hpp"
69 70
#include "dealer.hpp"
#include "router.hpp"
71 72
#include "xpub.hpp"
#include "xsub.hpp"
73

74 75 76 77 78
bool zmq::socket_base_t::check_tag ()
{
    return tag == 0xbaddecaf;
}

79
zmq::socket_base_t *zmq::socket_base_t::create (int type_, class ctx_t *parent_,
80
    uint32_t tid_, int sid_)
81 82 83 84 85
{
    socket_base_t *s = NULL;
    switch (type_) {

    case ZMQ_PAIR:
86
        s = new (std::nothrow) pair_t (parent_, tid_, sid_);
87 88
        break;
    case ZMQ_PUB:
89
        s = new (std::nothrow) pub_t (parent_, tid_, sid_);
90 91
        break;
    case ZMQ_SUB:
92
        s = new (std::nothrow) sub_t (parent_, tid_, sid_);
93 94
        break;
    case ZMQ_REQ:
95
        s = new (std::nothrow) req_t (parent_, tid_, sid_);
96 97
        break;
    case ZMQ_REP:
98
        s = new (std::nothrow) rep_t (parent_, tid_, sid_);
99
        break;
100 101
    case ZMQ_DEALER:
        s = new (std::nothrow) dealer_t (parent_, tid_, sid_);
102
        break;
103 104
    case ZMQ_ROUTER:
        s = new (std::nothrow) router_t (parent_, tid_, sid_);
105 106
        break;     
    case ZMQ_PULL:
107
        s = new (std::nothrow) pull_t (parent_, tid_, sid_);
108 109
        break;
    case ZMQ_PUSH:
110
        s = new (std::nothrow) push_t (parent_, tid_, sid_);
111
        break;
112
    case ZMQ_XPUB:
113
        s = new (std::nothrow) xpub_t (parent_, tid_, sid_);
114 115
        break;
    case ZMQ_XSUB:
116
        s = new (std::nothrow) xsub_t (parent_, tid_, sid_);
117
        break;
118 119 120 121
    default:
        errno = EINVAL;
        return NULL;
    }
122
    alloc_assert (s);
123 124 125
    return s;
}

126
zmq::socket_base_t::socket_base_t (ctx_t *parent_, uint32_t tid_, int sid_) :
Martin Sustrik's avatar
Martin Sustrik committed
127
    own_t (parent_, tid_),
128
    tag (0xbaddecaf),
129
    ctx_terminated (false),
130
    destroyed (false),
Martin Sustrik's avatar
Martin Sustrik committed
131
    last_tsc (0),
Martin Sustrik's avatar
Martin Sustrik committed
132
    ticks (0),
133
    rcvmore (false)
Martin Sustrik's avatar
Martin Sustrik committed
134
{
135
    options.socket_id = sid_;
136 137 138 139
}

zmq::socket_base_t::~socket_base_t ()
{
140
    zmq_assert (destroyed);
141 142
}

143
zmq::mailbox_t *zmq::socket_base_t::get_mailbox ()
144
{
145
    return &mailbox;
146 147 148 149 150 151 152 153 154 155 156
}

void zmq::socket_base_t::stop ()
{
    //  Called by ctx when it is terminated (zmq_term).
    //  'stop' command is sent from the threads that called zmq_term to
    //  the thread owning the socket. This way, blocking call in the
    //  owner thread can be interrupted.
    send_stop ();
}

157 158 159 160 161 162 163 164 165 166 167 168 169
int zmq::socket_base_t::parse_uri (const char *uri_,
                        std::string &protocol_, std::string &address_)
{
    zmq_assert (uri_ != NULL);

    std::string uri (uri_);
    std::string::size_type pos = uri.find ("://");
    if (pos == std::string::npos) {
        errno = EINVAL;
        return -1;
    }
    protocol_ = uri.substr (0, pos);
    address_ = uri.substr (pos + 3);
170
    
171 172 173 174 175 176 177
    if (protocol_.empty () || address_.empty ()) {
        errno = EINVAL;
        return -1;
    }
    return 0;
}

178 179 180 181
int zmq::socket_base_t::check_protocol (const std::string &protocol_)
{
    //  First check out whether the protcol is something we are aware of.
    if (protocol_ != "inproc" && protocol_ != "ipc" && protocol_ != "tcp" &&
Pieter Hintjens's avatar
Pieter Hintjens committed
182
          protocol_ != "pgm" && protocol_ != "epgm") {
183 184 185 186 187 188 189 190 191 192 193 194 195 196 197
        errno = EPROTONOSUPPORT;
        return -1;
    }

    //  If 0MQ is not compiled with OpenPGM, pgm and epgm transports
    //  are not avaialble.
#if !defined ZMQ_HAVE_OPENPGM
    if (protocol_ == "pgm" || protocol_ == "epgm") {
        errno = EPROTONOSUPPORT;
        return -1;
    }
#endif

    //  IPC transport is not available on Windows and OpenVMS.
#if defined ZMQ_HAVE_WINDOWS || defined ZMQ_HAVE_OPENVMS
198
    if (protocol_ == "ipc") {
199 200 201 202 203 204 205 206 207 208
        //  Unknown protocol.
        errno = EPROTONOSUPPORT;
        return -1;
    }
#endif

    //  Check whether socket type and transport protocol match.
    //  Specifically, multicast protocols can't be combined with
    //  bi-directional messaging patterns (socket types).
    if ((protocol_ == "pgm" || protocol_ == "epgm") &&
209 210
          options.type != ZMQ_PUB && options.type != ZMQ_SUB &&
          options.type != ZMQ_XPUB && options.type != ZMQ_XSUB) {
211 212 213 214 215 216 217 218
        errno = ENOCOMPATPROTO;
        return -1;
    }

    //  Protocol is available.
    return 0;
}

219
void zmq::socket_base_t::attach_pipe (pipe_t *pipe_, bool icanhasall_)
220
{
221 222 223 224
    //  First, register the pipe so that we can terminate it later on.
    pipe_->set_event_sink (this);
    pipes.push_back (pipe_);
    
225
    //  Let the derived socket type know about new pipe.
226
    xattach_pipe (pipe_, icanhasall_);
227 228 229 230 231

    //  If the socket is already being closed, ask any new pipes to terminate
    //  straight away.
    if (is_terminating ()) {
        register_term_acks (1);
232
        pipe_->terminate (false);
233
    }
234 235
}

236
int zmq::socket_base_t::setsockopt (int option_, const void *optval_,
237
    size_t optvallen_)
238
{
239
    if (unlikely (ctx_terminated)) {
240 241 242 243
        errno = ETERM;
        return -1;
    }

244 245 246 247 248 249 250 251
    //  First, check whether specific socket type overloads the option.
    int rc = xsetsockopt (option_, optval_, optvallen_);
    if (rc == 0 || errno != EINVAL)
        return rc;

    //  If the socket type doesn't support the option, pass it to
    //  the generic option parser.
    return options.setsockopt (option_, optval_, optvallen_);
252 253
}

254 255 256
int zmq::socket_base_t::getsockopt (int option_, void *optval_,
    size_t *optvallen_)
{
257
    if (unlikely (ctx_terminated)) {
258 259 260 261
        errno = ETERM;
        return -1;
    }

262
    if (option_ == ZMQ_RCVMORE) {
263
        if (*optvallen_ < sizeof (int)) {
264 265 266
            errno = EINVAL;
            return -1;
        }
267 268
        *((int*) optval_) = rcvmore ? 1 : 0;
        *optvallen_ = sizeof (int);
269 270 271
        return 0;
    }

272 273 274 275 276
    if (option_ == ZMQ_FD) {
        if (*optvallen_ < sizeof (fd_t)) {
            errno = EINVAL;
            return -1;
        }
277
        *((fd_t*) optval_) = mailbox.get_fd ();
278 279 280 281 282
        *optvallen_ = sizeof (fd_t);
        return 0;
    }

    if (option_ == ZMQ_EVENTS) {
283
        if (*optvallen_ < sizeof (int)) {
284 285 286
            errno = EINVAL;
            return -1;
        }
287
        int rc = process_commands (0, false);
288
        if (rc != 0 && (errno == EINTR || errno == ETERM))
289 290
            return -1;
        errno_assert (rc == 0);
291
        *((int*) optval_) = 0;
292
        if (has_out ())
293
            *((int*) optval_) |= ZMQ_POLLOUT;
294
        if (has_in ())
295 296
            *((int*) optval_) |= ZMQ_POLLIN;
        *optvallen_ = sizeof (int);
297 298 299
        return 0;
    }

300 301 302
    return options.getsockopt (option_, optval_, optvallen_);
}

303 304
int zmq::socket_base_t::bind (const char *addr_)
{
305
    if (unlikely (ctx_terminated)) {
306 307 308 309
        errno = ETERM;
        return -1;
    }

310 311 312 313 314
    //  Process pending commands, if any.
    int rc = process_commands (0, false);
    if (unlikely (rc != 0))
        return -1;

315
    //  Parse addr_ string.
316 317
    std::string protocol;
    std::string address;
318
    rc = parse_uri (addr_, protocol, address);
319 320
    if (rc != 0)
        return -1;
321

322
    rc = check_protocol (protocol);
323 324
    if (rc != 0)
        return -1;
325

Pieter Hintjens's avatar
Pieter Hintjens committed
326
    if (protocol == "inproc") {
327
        endpoint_t endpoint = {this, options};
328 329
        int rc = register_endpoint (addr_, endpoint);
        if (rc == 0) {
330 331
            // Save last endpoint URI
            options.last_endpoint.assign (addr_);
332 333
        }
        return rc;
334
    }
335

336 337 338
    if (protocol == "pgm" || protocol == "epgm") {
        //  For convenience's sake, bind can be used interchageable with
        //  connect for PGM and EPGM transports.
339
        return connect (addr_);
340 341 342 343 344 345 346 347 348
    }

    //  Remaining trasnports require to be run in an I/O thread, so at this
    //  point we'll choose one.
    io_thread_t *io_thread = choose_io_thread (options.affinity);
    if (!io_thread) {
        errno = EMTHREAD;
        return -1;
    }
349

350
    if (protocol == "tcp") {
351
        tcp_listener_t *listener = new (std::nothrow) tcp_listener_t (
352
            io_thread, this, options);
353
        alloc_assert (listener);
354
        int rc = listener->set_address (address.c_str ());
355 356
        if (rc != 0) {
            delete listener;
357
            monitor_event (ZMQ_EVENT_BIND_FAILED, addr_, zmq_errno());
358
            return -1;
359
        }
360

361
        // Save last endpoint URI
362 363
        listener->get_address (options.last_endpoint);

364
        add_endpoint (addr_, (own_t *) listener);
365 366 367
        return 0;
    }

368
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
369 370 371 372
    if (protocol == "ipc") {
        ipc_listener_t *listener = new (std::nothrow) ipc_listener_t (
            io_thread, this, options);
        alloc_assert (listener);
373
        int rc = listener->set_address (address.c_str ());
374 375
        if (rc != 0) {
            delete listener;
376
            monitor_event (ZMQ_EVENT_BIND_FAILED, addr_, zmq_errno());
377 378
            return -1;
        }
379

380
        // Save last endpoint URI
381 382
        listener->get_address (options.last_endpoint);

383
        add_endpoint (addr_, (own_t *) listener);
384
        return 0;
385
    }
386
#endif
387

388
    zmq_assert (false);
Martin Sustrik's avatar
Martin Sustrik committed
389
    return -1;
390 391
}

392
int zmq::socket_base_t::connect (const char *addr_)
393
{
394
    if (unlikely (ctx_terminated)) {
395 396 397 398
        errno = ETERM;
        return -1;
    }

399 400 401 402 403
    //  Process pending commands, if any.
    int rc = process_commands (0, false);
    if (unlikely (rc != 0))
        return -1;

malosek's avatar
malosek committed
404
    //  Parse addr_ string.
405 406
    std::string protocol;
    std::string address;
407
    rc = parse_uri (addr_, protocol, address);
408 409
    if (rc != 0)
        return -1;
malosek's avatar
malosek committed
410

411
    rc = check_protocol (protocol);
412 413
    if (rc != 0)
        return -1;
malosek's avatar
malosek committed
414

Pieter Hintjens's avatar
Pieter Hintjens committed
415
    if (protocol == "inproc") {
416

417 418 419 420
        //  TODO: inproc connect is specific with respect to creating pipes
        //  as there's no 'reconnect' functionality implemented. Once that
        //  is in place we should follow generic pipe creation algorithm.

421 422 423
        //  Find the peer endpoint.
        endpoint_t peer = find_endpoint (addr_);
        if (!peer.socket)
424 425
            return -1;

426
        // The total HWM for an inproc connection should be the sum of
427
        // the binder's HWM and the connector's HWM.
428 429 430 431
        int  sndhwm;
        int  rcvhwm;
        if (options.sndhwm == 0 || peer.options.rcvhwm == 0)
            sndhwm = 0;
432
        else
433 434 435 436 437
            sndhwm = options.sndhwm + peer.options.rcvhwm;
        if (options.rcvhwm == 0 || peer.options.sndhwm == 0)
            rcvhwm = 0;
        else
            rcvhwm = options.rcvhwm + peer.options.sndhwm;
438

439 440 441 442
        //  Create a bi-directional pipe to connect the peers.
        object_t *parents [2] = {this, peer.socket};
        pipe_t *pipes [2] = {NULL, NULL};
        int hwms [2] = {sndhwm, rcvhwm};
443
        bool delays [2] = {options.delay_on_disconnect, options.delay_on_close};
444 445
        int rc = pipepair (parents, pipes, hwms, delays);
        errno_assert (rc == 0);
446

447
        //  Attach local end of the pipe to this socket object.
448
        attach_pipe (pipes [0]);
449

450 451 452 453
        //  If required, send the identity of the local socket to the peer.
        if (options.send_identity) {
            msg_t id;
            rc = id.init_size (options.identity_size);
454
            errno_assert (rc == 0);
455 456 457 458
            memcpy (id.data (), options.identity, options.identity_size);
            id.set_flags (msg_t::identity);
            bool written = pipes [0]->write (&id);
            zmq_assert (written);
459
            pipes [0]->flush ();
460 461
        }

462 463 464 465
        //  If required, send the identity of the peer to the local socket.
        if (peer.options.send_identity) {
            msg_t id;
            rc = id.init_size (peer.options.identity_size);
466
            errno_assert (rc == 0);
467 468 469 470 471 472 473
            memcpy (id.data (), peer.options.identity, peer.options.identity_size);
            id.set_flags (msg_t::identity);
            bool written = pipes [1]->write (&id);
            zmq_assert (written);
            pipes [1]->flush ();
        }

474 475
        //  Attach remote end of the pipe to the peer socket. Note that peer's
        //  seqnum was incremented in find_endpoint function. We don't need it
476
        //  increased here.
477
        send_bind (peer.socket, pipes [1], false);
478

479 480
        // Save last endpoint URI
        options.last_endpoint.assign (addr_);
481

482 483 484
        return 0;
    }

485 486 487 488 489 490 491
    //  Choose the I/O thread to run the session in.
    io_thread_t *io_thread = choose_io_thread (options.affinity);
    if (!io_thread) {
        errno = EMTHREAD;
        return -1;
    }

492
    address_t *paddr = new (std::nothrow) address_t (protocol, address);
493
    alloc_assert (paddr);
494 495 496 497

    //  Resolve address (if needed by the protocol)
    if (protocol == "tcp") {
        paddr->resolved.tcp_addr = new (std::nothrow) tcp_address_t ();
498
        alloc_assert (paddr->resolved.tcp_addr);
499 500 501 502 503 504 505
        int rc = paddr->resolved.tcp_addr->resolve (
            address.c_str (), false, options.ipv4only ? true : false);
        if (rc != 0) {
            delete paddr;
            return -1;
        }
    }
506
#if !defined ZMQ_HAVE_WINDOWS && !defined ZMQ_HAVE_OPENVMS
507 508
    else if(protocol == "ipc") {
        paddr->resolved.ipc_addr = new (std::nothrow) ipc_address_t ();
509
        alloc_assert (paddr->resolved.ipc_addr);
510 511 512 513 514 515
        int rc = paddr->resolved.ipc_addr->resolve (address.c_str ());
        if (rc != 0) {
            delete paddr;
            return -1;
        }
    }
516 517 518 519 520 521 522 523 524 525 526
#endif
#ifdef ZMQ_HAVE_OPENPGM
    if (protocol == "pgm" || protocol == "epgm") {
        struct pgm_addrinfo_t *res = NULL;
        uint16_t port_number = 0;
        int rc = pgm_socket_t::init_address(address.c_str(), &res, &port_number);
        if (res != NULL)
            pgm_freeaddrinfo (res);
        if (rc != 0 || port_number == 0)
            return -1;
    }
527
#endif
528
    //  Create session.
529
    session_base_t *session = session_base_t::create (io_thread, true, this,
530
        options, paddr);
531
    errno_assert (session);
532

533 534 535 536 537
    //  PGM does not support subscription forwarding; ask for all data to be
    //  sent to this pipe.
    bool icanhasall = false;
    if (protocol == "pgm" || protocol == "epgm")
        icanhasall = true;
538

539
    if (options.delay_attach_on_connect != 1 || icanhasall) {
540 541 542 543 544 545 546
        //  Create a bi-directional pipe.
        object_t *parents [2] = {this, session};
        pipe_t *pipes [2] = {NULL, NULL};
        int hwms [2] = {options.sndhwm, options.rcvhwm};
        bool delays [2] = {options.delay_on_disconnect, options.delay_on_close};
        rc = pipepair (parents, pipes, hwms, delays);
        errno_assert (rc == 0);
547

548
        //  Attach local end of the pipe to the socket object.
549
        attach_pipe (pipes [0], icanhasall);
550

551 552 553
        //  Attach remote end of the pipe to the session object later on.
        session->attach_pipe (pipes [1]);
    }
Martin Sustrik's avatar
Martin Sustrik committed
554

555
    // Save last endpoint URI
556 557
    paddr->to_string (options.last_endpoint);

558 559 560 561 562 563
    add_endpoint (addr_, (own_t *) session);
    return 0;
}

void zmq::socket_base_t::add_endpoint (const char *addr_, own_t *endpoint_)
{
564
    //  Activate the session. Make it a child of this socket.
565
    launch_child (endpoint_);
566
    endpoints.insert (endpoints_t::value_type (std::string (addr_), endpoint_));
567 568 569 570 571 572 573 574 575
}

int zmq::socket_base_t::term_endpoint (const char *addr_)
{
    //  Check whether the library haven't been shut down yet.
    if (unlikely (ctx_terminated)) {
        errno = ETERM;
        return -1;
    }
malosek's avatar
malosek committed
576

577
    //  Check whether endpoint address passed to the function is valid.
578 579 580 581 582
    if (unlikely (!addr_)) {
        errno = EINVAL;
        return -1;
    }

583 584 585 586 587 588
    //  Process pending commands, if any, since there could be pending unprocessed process_own()'s
    //  (from launch_child() for example) we're asked to terminate now.
    int rc = process_commands (0, false);
    if (unlikely (rc != 0))
        return -1;

589 590 591 592 593 594 595 596
    //  Find the endpoints range (if any) corresponding to the addr_ string.
    std::pair <endpoints_t::iterator, endpoints_t::iterator> range = endpoints.equal_range (std::string (addr_));
    if (range.first == range.second)
        return -1;

    for (endpoints_t::iterator it = range.first; it != range.second; ++it)
        term_child (it->second);
    endpoints.erase (range.first, range.second);
597
    return 0;
598 599
}

600
int zmq::socket_base_t::send (msg_t *msg_, int flags_)
601
{
602
    //  Check whether the library haven't been shut down yet.
603
    if (unlikely (ctx_terminated)) {
604 605 606 607
        errno = ETERM;
        return -1;
    }

608
    //  Check whether message passed to the function is valid.
609
    if (unlikely (!msg_ || !msg_->check ())) {
610 611 612 613
        errno = EFAULT;
        return -1;
    }

614
    //  Process pending commands, if any.
615
    int rc = process_commands (0, true);
616
    if (unlikely (rc != 0))
617 618
        return -1;

619 620 621
    //  Clear any user-visible flags that are set on the message.
    msg_->reset_flags (msg_t::more);

622
    //  At this point we impose the flags on the message.
623
    if (flags_ & ZMQ_SNDMORE)
624
        msg_->set_flags (msg_t::more);
625

Martin Sustrik's avatar
Martin Sustrik committed
626
    //  Try to send the message.
627
    rc = xsend (msg_, flags_);
628 629
    if (rc == 0)
        return 0;
630 631
    if (unlikely (errno != EAGAIN))
        return -1;
Martin Sustrik's avatar
Martin Sustrik committed
632

633
    //  In case of non-blocking send we'll simply propagate
634 635
    //  the error - including EAGAIN - up the stack.
    if (flags_ & ZMQ_DONTWAIT || options.sndtimeo == 0)
Martin Sustrik's avatar
Martin Sustrik committed
636 637
        return -1;

638 639 640 641 642
    //  Compute the time when the timeout should occur.
    //  If the timeout is infite, don't care. 
    int timeout = options.sndtimeo;
    uint64_t end = timeout < 0 ? 0 : (clock.now_ms () + timeout);

643 644
    //  Oops, we couldn't send the message. Wait for the next
    //  command, process it and try to send the message again.
645 646 647
    //  If timeout is reached in the meantime, return EAGAIN.
    while (true) {
        if (unlikely (process_commands (timeout, false) != 0))
648
            return -1;
649
        rc = xsend (msg_, flags_);
650 651 652 653 654
        if (rc == 0)
            break;
        if (unlikely (errno != EAGAIN))
            return -1;
        if (timeout > 0) {
655
            timeout = (int) (end - clock.now_ms ());
656 657 658 659 660
            if (timeout <= 0) {
                errno = EAGAIN;
                return -1;
            }
        }
661
    }
Martin Sustrik's avatar
Martin Sustrik committed
662
    return 0;
663 664
}

665
int zmq::socket_base_t::recv (msg_t *msg_, int flags_)
666
{
667
    //  Check whether the library haven't been shut down yet.
668
    if (unlikely (ctx_terminated)) {
669 670 671 672
        errno = ETERM;
        return -1;
    }

673
    //  Check whether message passed to the function is valid.
674
    if (unlikely (!msg_ || !msg_->check ())) {
675 676 677 678
        errno = EFAULT;
        return -1;
    }

679
    //  Get the message.
680
    int rc = xrecv (msg_, flags_);
681 682
    if (unlikely (rc != 0 && errno != EAGAIN))
        return -1;
683 684 685 686 687 688 689 690

    //  Once every inbound_poll_rate messages check for signals and process
    //  incoming commands. This happens only if we are not polling altogether
    //  because there are messages available all the time. If poll occurs,
    //  ticks is set to zero and thus we avoid this code.
    //
    //  Note that 'recv' uses different command throttling algorithm (the one
    //  described above) from the one used by 'send'. This is because counting
Martin Sustrik's avatar
Martin Sustrik committed
691
    //  ticks is more efficient than doing RDTSC all the time.
692
    if (++ticks == inbound_poll_rate) {
693
        if (unlikely (process_commands (0, false) != 0))
694
            return -1;
695 696 697 698
        ticks = 0;
    }

    //  If we have the message, return immediately.
699
    if (rc == 0) {
700
        extract_flags (msg_);
701
        return 0;
702
    }
703

Martin Sustrik's avatar
Martin Sustrik committed
704
    //  If the message cannot be fetched immediately, there are two scenarios.
705 706 707
    //  For non-blocking recv, commands are processed in case there's an
    //  activate_reader command already waiting int a command pipe.
    //  If it's not, return EAGAIN.
708 709
    if (flags_ & ZMQ_DONTWAIT || options.rcvtimeo == 0) {
        if (unlikely (process_commands (0, false) != 0))
710
            return -1;
711
        ticks = 0;
712 713

        rc = xrecv (msg_, flags_);
714 715
        if (rc < 0)
            return rc;
716
        extract_flags (msg_);
717
        return 0;
Martin Sustrik's avatar
Martin Sustrik committed
718 719
    }

720 721 722 723 724
    //  Compute the time when the timeout should occur.
    //  If the timeout is infite, don't care. 
    int timeout = options.rcvtimeo;
    uint64_t end = timeout < 0 ? 0 : (clock.now_ms () + timeout);

725 726
    //  In blocking scenario, commands are processed over and over again until
    //  we are able to fetch a message.
727
    bool block = (ticks != 0);
728 729
    while (true) {
        if (unlikely (process_commands (block ? timeout : 0, false) != 0))
730
            return -1;
731
        rc = xrecv (msg_, flags_);
732 733 734 735 736 737
        if (rc == 0) {
            ticks = 0;
            break;
        }
        if (unlikely (errno != EAGAIN))
            return -1;
738
        block = true;
739
        if (timeout > 0) {
740
            timeout = (int) (end - clock.now_ms ());
741 742 743 744 745
            if (timeout <= 0) {
                errno = EAGAIN;
                return -1;
            }
        }
746
    }
747

748
    extract_flags (msg_);
749
    return 0;
750 751 752 753
}

int zmq::socket_base_t::close ()
{
754 755 756
    //  Mark the socket as dead
    tag = 0xdeadbeef;
    
757 758 759 760
    //  Transfer the ownership of the socket from this application thread
    //  to the reaper thread which will take care of the rest of shutdown
    //  process.
    send_reap (this);
761

762 763 764
    return 0;
}

765 766 767 768 769 770 771 772 773 774
bool zmq::socket_base_t::has_in ()
{
    return xhas_in ();
}

bool zmq::socket_base_t::has_out ()
{
    return xhas_out ();
}

775
void zmq::socket_base_t::start_reaping (poller_t *poller_)
776
{
777
    //  Plug the socket to the reaper thread.
778 779 780
    poller = poller_;
    handle = poller->add_fd (mailbox.get_fd (), this);
    poller->set_pollin (handle);
781 782 783 784 785

    //  Initialise the termination and check whether it can be deallocated
    //  immediately.
    terminate ();
    check_destroy ();
Martin Hurton's avatar
Martin Hurton committed
786 787
}

788
int zmq::socket_base_t::process_commands (int timeout_, bool throttle_)
Martin Sustrik's avatar
Martin Sustrik committed
789
{
790
    int rc;
791
    command_t cmd;
792 793 794 795
    if (timeout_ != 0) {

        //  If we are asked to wait, simply ask mailbox to wait.
        rc = mailbox.recv (&cmd, timeout_);
796 797
    }
    else {
798

799 800 801
        //  If we are asked not to wait, check whether we haven't processed
        //  commands recently, so that we can throttle the new commands.

Martin Sustrik's avatar
Martin Sustrik committed
802 803 804
        //  Get the CPU's tick counter. If 0, the counter is not available.
        uint64_t tsc = zmq::clock_t::rdtsc ();

805 806 807 808 809 810
        //  Optimised version of command processing - it doesn't have to check
        //  for incoming commands each time. It does so only if certain time
        //  elapsed since last command processing. Command delay varies
        //  depending on CPU speed: It's ~1ms on 3GHz CPU, ~2ms on 1.5GHz CPU
        //  etc. The optimisation makes sense only on platforms where getting
        //  a timestamp is a very cheap operation (tens of nanoseconds).
Martin Sustrik's avatar
Martin Sustrik committed
811 812
        if (tsc && throttle_) {

Martin Sustrik's avatar
Martin Sustrik committed
813 814 815
            //  Check whether TSC haven't jumped backwards (in case of migration
            //  between CPU cores) and whether certain time have elapsed since
            //  last command processing. If it didn't do nothing.
Martin Sustrik's avatar
Martin Sustrik committed
816
            if (tsc >= last_tsc && tsc - last_tsc <= max_command_delay)
817
                return 0;
Martin Sustrik's avatar
Martin Sustrik committed
818
            last_tsc = tsc;
819 820 821
        }

        //  Check whether there are any commands pending for this thread.
822
        rc = mailbox.recv (&cmd, 0);
823
    }
Martin Sustrik's avatar
Martin Sustrik committed
824

825
    //  Process all the commands available at the moment.
826 827 828 829 830 831
    while (true) {
        if (rc == -1 && errno == EAGAIN)
            break;
        if (rc == -1 && errno == EINTR)
            return -1;
        errno_assert (rc == 0);
832
        cmd.destination->process_command (cmd);
833
        rc = mailbox.recv (&cmd, 0);
834 835
     }

836
    if (ctx_terminated) {
837 838
        errno = ETERM;
        return -1;
839
    }
840 841

    return 0;
Martin Sustrik's avatar
Martin Sustrik committed
842 843
}

844
void zmq::socket_base_t::process_stop ()
Martin Sustrik's avatar
Martin Sustrik committed
845
{
846
    //  Here, someone have called zmq_term while the socket was still alive.
847
    //  We'll remember the fact so that any blocking call is interrupted and any
848 849
    //  further attempt to use the socket will return ETERM. The user is still
    //  responsible for calling zmq_close on the socket though!
850
    ctx_terminated = true;
Martin Sustrik's avatar
Martin Sustrik committed
851 852
}

853
void zmq::socket_base_t::process_bind (pipe_t *pipe_)
Martin Sustrik's avatar
Martin Sustrik committed
854
{
855
    attach_pipe (pipe_);
Martin Sustrik's avatar
Martin Sustrik committed
856 857
}

858
void zmq::socket_base_t::process_term (int linger_)
859
{
860 861 862 863 864
    //  Unregister all inproc endpoints associated with this socket.
    //  Doing this we make sure that no new pipes from other sockets (inproc)
    //  will be initiated.
    unregister_endpoints (this);

865 866
    //  Ask all attached pipes to terminate.
    for (pipes_t::size_type i = 0; i != pipes.size (); ++i)
867
        pipes [i]->terminate (false);
Martin Sustrik's avatar
Martin Sustrik committed
868
    register_term_acks ((int) pipes.size ());
869

870
    //  Continue the termination process immediately.
871
    own_t::process_term (linger_);
872 873
}

874 875 876 877 878
void zmq::socket_base_t::process_destroy ()
{
    destroyed = true;
}

879 880 881 882 883 884 885 886 887 888 889 890
int zmq::socket_base_t::xsetsockopt (int option_, const void *optval_,
    size_t optvallen_)
{
    errno = EINVAL;
    return -1;
}

bool zmq::socket_base_t::xhas_out ()
{
    return false;
}

891
int zmq::socket_base_t::xsend (msg_t *msg_, int flags_)
892 893 894 895 896 897 898 899 900 901
{
    errno = ENOTSUP;
    return -1;
}

bool zmq::socket_base_t::xhas_in ()
{
    return false;
}

902
int zmq::socket_base_t::xrecv (msg_t *msg_, int flags_)
903 904 905 906 907
{
    errno = ENOTSUP;
    return -1;
}

908 909 910 911 912 913 914 915 916
void zmq::socket_base_t::xread_activated (pipe_t *pipe_)
{
    zmq_assert (false);
}
void zmq::socket_base_t::xwrite_activated (pipe_t *pipe_)
{
    zmq_assert (false);
}

917 918 919 920 921
void zmq::socket_base_t::xhiccuped (pipe_t *pipe_)
{
    zmq_assert (false);
}

922 923
void zmq::socket_base_t::in_event ()
{
924 925 926 927 928
    //  This function is invoked only once the socket is running in the context
    //  of the reaper thread. Process any commands from other threads/sockets
    //  that may be available at the moment. Ultimately, the socket will
    //  be destroyed.
    process_commands (0, false);
929 930 931 932 933 934 935 936 937 938 939 940
    check_destroy ();
}

void zmq::socket_base_t::out_event ()
{
    zmq_assert (false);
}

void zmq::socket_base_t::timer_event (int id_)
{
    zmq_assert (false);
}
941

942 943
void zmq::socket_base_t::check_destroy ()
{
944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959
    //  If the object was already marked as destroyed, finish the deallocation.
    if (destroyed) {

        //  Remove the socket from the reaper's poller.
        poller->rm_fd (handle);

        //  Remove the socket from the context.
        destroy_socket (this);

        //  Notify the reaper about the fact.
        send_reaped ();

        //  Deallocate.
        own_t::process_destroy ();
    }
}
960 961 962 963 964 965 966 967 968 969 970

void zmq::socket_base_t::read_activated (pipe_t *pipe_)
{
    xread_activated (pipe_);
}

void zmq::socket_base_t::write_activated (pipe_t *pipe_)
{
    xwrite_activated (pipe_);
}

971 972 973 974 975
void zmq::socket_base_t::hiccuped (pipe_t *pipe_)
{
    xhiccuped (pipe_);
}

976 977 978 979 980 981 982 983 984 985 986 987
void zmq::socket_base_t::terminated (pipe_t *pipe_)
{
    //  Notify the specific socket type about the pipe termination.
    xterminated (pipe_);

    //  Remove the pipe from the list of attached pipes and confirm its
    //  termination if we are already shutting down.
    pipes.erase (pipe_);
    if (is_terminating ())
        unregister_term_ack ();
}

988 989
void zmq::socket_base_t::extract_flags (msg_t *msg_)
{
990
    //  Test whether IDENTITY flag is valid for this socket type.
991
    if (unlikely (msg_->flags () & msg_t::identity))
992 993 994
        zmq_assert (options.recv_identity);
  
    //  Remove MORE flag.
995 996
    rcvmore = msg_->flags () & msg_t::more ? true : false;
}
997 998 999

void zmq::socket_base_t::monitor_event (int event_, ...)
{
1000 1001 1002 1003
    va_list args;
    va_start (args, event_);
    get_ctx ()->monitor_event (this, event_, args);
    va_end (args);
1004
}