1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
/*
Copyright (c) 2007-2009 FastMQ Inc.
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the Lesser GNU General Public License as published by
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
Lesser GNU General Public License for more details.
You should have received a copy of the Lesser GNU General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include <algorithm>
#include "../include/zmq.h"
#include "socket_base.hpp"
#include "app_thread.hpp"
#include "err.hpp"
#include "zmq_listener.hpp"
#include "io_thread.hpp"
zmq::socket_base_t::socket_base_t (app_thread_t *parent_) :
object_t (parent_),
pending_term_acks (0),
app_thread (parent_)
{
}
zmq::socket_base_t::~socket_base_t ()
{
while (true) {
// On third pass of the loop there should be no more I/O objects
// because all connecters and listerners were destroyed during
// the first pass and all engines delivered by delayed 'own' commands
// are destroyed during the second pass.
if (io_objects.empty () && !pending_term_acks)
break;
// Send termination request to all associated I/O objects.
for (io_objects_t::size_type i = 0; i != io_objects.size (); i++)
send_term (io_objects [i]);
// Move the objects to the list of pending term acks.
pending_term_acks += io_objects.size ();
io_objects.clear ();
// Process commands till we get all the termination acknowledgements.
while (pending_term_acks)
app_thread->process_commands (true);
}
}
int zmq::socket_base_t::bind (const char *addr_, struct zmq_opts *opts_)
{
uint64_t taskset = opts_ ? opts_->taskset : 0;
object_t *listener = new zmq_listener_t (choose_io_thread (taskset), this);
send_plug (listener);
send_own (this, listener);
return 0;
}
int zmq::socket_base_t::connect (const char *addr_, struct zmq_opts *opts_)
{
zmq_assert (false);
}
int zmq::socket_base_t::subscribe (const char *criteria_)
{
zmq_assert (false);
}
int zmq::socket_base_t::send (struct zmq_msg *msg_, int flags_)
{
zmq_assert (false);
}
int zmq::socket_base_t::flush ()
{
zmq_assert (false);
}
int zmq::socket_base_t::recv (struct zmq_msg *msg_, int flags_)
{
zmq_assert (false);
}
int zmq::socket_base_t::close ()
{
app_thread->remove_socket (this);
delete this;
return 0;
}
void zmq::socket_base_t::process_own (object_t *object_)
{
io_objects.push_back (object_);
}
void zmq::socket_base_t::process_term_req (object_t *object_)
{
// If I/O object is well and alive ask it to terminate.
// TODO: Following find may produce an unacceptable jitter in
// C10K-style applications. If so, use set instead of vector.
io_objects_t::iterator it = std::find (io_objects.begin (),
io_objects.end (), object_);
if (it != io_objects.end ()) {
pending_term_acks++;
io_objects.erase (it);
send_term (object_);
}
// If not found, we assume that termination request was already sent to
// the object so we can sagely ignore the request.
}
void zmq::socket_base_t::process_term_ack ()
{
zmq_assert (pending_term_acks);
pending_term_acks--;
}
|
