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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
/*
Copyright (c) 2007-2010 iMatix Corporation
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/>.
*/
#ifndef __ZMQ_CTX_HPP_INCLUDED__
#define __ZMQ_CTX_HPP_INCLUDED__
#include <map>
#include <vector>
#include <string>
#include "../include/zmq.h"
#include "signaler.hpp"
#include "semaphore.hpp"
#include "ypipe.hpp"
#include "array.hpp"
#include "config.hpp"
#include "mutex.hpp"
#include "stdint.hpp"
#include "thread.hpp"
namespace zmq
{
// Context object encapsulates all the global state associated with
// the library.
class ctx_t
{
public:
// Create the context object. The argument specifies the size
// of I/O thread pool to create.
ctx_t (uint32_t io_threads_);
// This function is called when user invokes zmq_term. If there are
// no more sockets open it'll cause all the infrastructure to be shut
// down. If there are open sockets still, the deallocation happens
// after the last one is closed.
int terminate ();
// Create a socket.
class socket_base_t *create_socket (int type_);
// Make socket a zombie.
void zombify_socket (socket_base_t *socket_);
// Send command to the destination slot.
void send_command (uint32_t slot_, const command_t &command_);
// Receive command from the source slot.
bool recv_command (uint32_t slot_, command_t *command_, bool block_);
// Returns the I/O thread that is the least busy at the moment.
// Taskset specifies which I/O threads are eligible (0 = all).
class io_thread_t *choose_io_thread (uint64_t taskset_);
// Management of inproc endpoints.
int register_endpoint (const char *addr_, class socket_base_t *socket_);
void unregister_endpoints (class socket_base_t *socket_);
class socket_base_t *find_endpoint (const char *addr_);
// Logging.
void log (zmq_msg_t *msg_);
private:
~ctx_t ();
// Sockets belonging to this context.
typedef array_t <socket_base_t> sockets_t;
sockets_t sockets;
// List of sockets that were already closed but not yet deallocated.
// These sockets still have some pipes and I/O objects attached.
typedef std::vector <socket_base_t*> zombies_t;
zombies_t zombies;
// List of unused slots.
typedef std::vector <uint32_t> emtpy_slots_t;
emtpy_slots_t empty_slots;
// If true, shutdown thread wants to be informed when there are no
// more open sockets. Do so by posting no_sockets_sync semaphore.
// Note that this variable is synchronised by slot_sync mutex.
bool no_sockets_notify;
// Object used by zmq_term to wait while all the sockets are closed
// by different application threads.
semaphore_t no_sockets_sync;
// Synchronisation of accesses to global slot-related data:
// sockets, zombies, empty_slots, terminated. It also synchronises
// access to zombie sockets as such (as oposed to slots) and provides
// a memory barrier to ensure that all CPU cores see the same data.
mutex_t slot_sync;
// This function attempts to deallocate as many zombie sockets as
// possible. It must be called within a slot_sync critical section.
void dezombify ();
// I/O threads.
typedef std::vector <class io_thread_t*> io_threads_t;
io_threads_t io_threads;
// Array of pointers to signalers for both application and I/O threads.
uint32_t slot_count;
signaler_t **slots;
// List of inproc endpoints within this context.
typedef std::map <std::string, class socket_base_t*> endpoints_t;
endpoints_t endpoints;
// Synchronisation of access to the list of inproc endpoints.
mutex_t endpoints_sync;
// PUB socket for logging. The socket is shared among all the threads,
// thus it is synchronised by a mutex.
class socket_base_t *log_socket;
mutex_t log_sync;
ctx_t (const ctx_t&);
void operator = (const ctx_t&);
};
}
#endif
|
