blob: f14c578bff0f53574fd5f5c45568e40f022681d3 (
plain)
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
|
/*
Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
This file is part of 0MQ.
0MQ is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser 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
GNU Lesser General Public License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_OWN_HPP_INCLUDED__
#define __ZMQ_OWN_HPP_INCLUDED__
#include <set>
#include <algorithm>
#include "object.hpp"
#include "options.hpp"
#include "atomic_counter.hpp"
#include "stdint.hpp"
namespace zmq
{
class ctx_t;
class io_thread_t;
// Base class for objects forming a part of ownership hierarchy.
// It handles initialisation and destruction of such objects.
class own_t : public object_t
{
public:
// Note that the owner is unspecified in the constructor.
// It'll be supplied later on when the object is plugged in.
// The object is not living within an I/O thread. It has it's own
// thread outside of 0MQ infrastructure.
own_t (zmq::ctx_t *parent_, uint32_t tid_);
// The object is living within I/O thread.
own_t (zmq::io_thread_t *io_thread_, const options_t &options_);
// When another owned object wants to send command to this object
// it calls this function to let it know it should not shut down
// before the command is delivered.
void inc_seqnum ();
// Use following two functions to wait for arbitrary events before
// terminating. Just add number of events to wait for using
// register_tem_acks functions. When event occurs, call
// remove_term_ack. When number of pending acks reaches zero
// object will be deallocated.
void register_term_acks (int count_);
void unregister_term_ack ();
protected:
// Launch the supplied object and become its owner.
void launch_child (own_t *object_);
// Terminate owned object
void term_child (own_t *object_);
// Ask owner object to terminate this object. It may take a while
// while actual termination is started. This function should not be
// called more than once.
void terminate ();
// Returns true if the object is in process of termination.
bool is_terminating ();
// Derived object destroys own_t. There's no point in allowing
// others to invoke the destructor. At the same time, it has to be
// virtual so that generic own_t deallocation mechanism destroys
// specific type of the owned object correctly.
virtual ~own_t ();
// Term handler is protocted rather than private so that it can
// be intercepted by the derived class. This is useful to add custom
// steps to the beginning of the termination process.
void process_term (int linger_);
// A place to hook in when phyicallal destruction of the object
// is to be delayed.
virtual void process_destroy ();
// Socket options associated with this object.
options_t options;
private:
// Set owner of the object
void set_owner (own_t *owner_);
// Handlers for incoming commands.
void process_own (own_t *object_);
void process_term_req (own_t *object_);
void process_term_ack ();
void process_seqnum ();
// Check whether all the peding term acks were delivered.
// If so, deallocate this object.
void check_term_acks ();
// True if termination was already initiated. If so, we can destroy
// the object if there are no more child objects or pending term acks.
bool terminating;
// Sequence number of the last command sent to this object.
atomic_counter_t sent_seqnum;
// Sequence number of the last command processed by this object.
uint64_t processed_seqnum;
// Socket owning this object. It's responsible for shutting down
// this object.
own_t *owner;
// List of all objects owned by this socket. We are responsible
// for deallocating them before we quit.
typedef std::set <own_t*> owned_t;
owned_t owned;
// Number of events we have to get before we can destroy the object.
int term_acks;
own_t (const own_t&);
const own_t &operator = (const own_t&);
};
}
#endif
|
