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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
/*
Copyright (c) 2009-2012 250bpm s.r.o.
Copyright (c) 2007-2009 iMatix Corporation
Copyright (c) 2012 Lucina & Associates
Copyright (c) 2010-2011 Miru Limited
Copyright (c) 2007-2011 Other contributors as noted in the AUTHORS file
This file is part of Crossroads project.
Crossroads 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.
Crossroads 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/>.
*/
#include "platform.hpp"
#if defined XS_HAVE_OPENPGM
#ifdef XS_HAVE_WINDOWS
#include "windows.hpp"
#endif
#include <stdlib.h>
#include "io_thread.hpp"
#include "pgm_sender.hpp"
#include "session_base.hpp"
#include "err.hpp"
#include "wire.hpp"
#include "stdint.hpp"
xs::pgm_sender_t::pgm_sender_t (io_thread_t *parent_,
const options_t &options_) :
io_object_t (parent_),
has_tx_timer (false),
has_rx_timer (false),
encoder (0),
pgm_socket (false, options_),
options (options_),
out_buffer (NULL),
out_buffer_size (0),
write_size (0)
{
}
int xs::pgm_sender_t::init (bool udp_encapsulation_, const char *network_)
{
int rc = pgm_socket.init (udp_encapsulation_, network_);
if (rc != 0)
return rc;
out_buffer_size = pgm_socket.get_max_tsdu_size ();
out_buffer = (unsigned char*) malloc (out_buffer_size);
alloc_assert (out_buffer);
return rc;
}
void xs::pgm_sender_t::plug (io_thread_t *io_thread_, session_base_t *session_)
{
// Alocate 2 fds for PGM socket.
fd_t downlink_socket_fd = retired_fd;
fd_t uplink_socket_fd = retired_fd;
fd_t rdata_notify_fd = retired_fd;
fd_t pending_notify_fd = retired_fd;
encoder.set_session (session_);
// Fill fds from PGM transport and add them to the poller.
pgm_socket.get_sender_fds (&downlink_socket_fd, &uplink_socket_fd,
&rdata_notify_fd, &pending_notify_fd);
handle = add_fd (downlink_socket_fd);
uplink_handle = add_fd (uplink_socket_fd);
rdata_notify_handle = add_fd (rdata_notify_fd);
pending_notify_handle = add_fd (pending_notify_fd);
// Set POLLIN. We wont never want to stop polling for uplink = we never
// want to stop porocess NAKs.
set_pollin (uplink_handle);
set_pollin (rdata_notify_handle);
set_pollin (pending_notify_handle);
// Set POLLOUT for downlink_socket_handle.
set_pollout (handle);
// PGM is not able to pass subscriptions upstream, thus we have no idea
// what messages are peers interested in. Because of that we have to
// subscribe for all the messages.
msg_t msg;
msg.init_size (1);
*(unsigned char*) msg.data () = 1;
int rc = session_->write (&msg);
errno_assert (rc == 0);
session_->flush ();
}
void xs::pgm_sender_t::unplug ()
{
if (has_rx_timer) {
cancel_timer (rx_timer_id);
has_rx_timer = false;
}
if (has_tx_timer) {
cancel_timer (tx_timer_id);
has_tx_timer = false;
}
rm_fd (handle);
rm_fd (uplink_handle);
rm_fd (rdata_notify_handle);
rm_fd (pending_notify_handle);
encoder.set_session (NULL);
}
void xs::pgm_sender_t::terminate ()
{
unplug ();
delete this;
}
void xs::pgm_sender_t::activate_out ()
{
set_pollout (handle);
out_event (retired_fd);
}
void xs::pgm_sender_t::activate_in ()
{
xs_assert (false);
}
xs::pgm_sender_t::~pgm_sender_t ()
{
if (out_buffer) {
free (out_buffer);
out_buffer = NULL;
}
}
void xs::pgm_sender_t::in_event (fd_t fd_)
{
if (has_rx_timer) {
cancel_timer (rx_timer_id);
has_rx_timer = false;
}
// In-event on sender side means NAK or SPMR receiving from some peer.
pgm_socket.process_upstream ();
if (errno == ENOMEM || errno == EBUSY) {
const long timeout = pgm_socket.get_rx_timeout ();
add_timer (timeout, rx_timer_id);
has_rx_timer = true;
}
}
void xs::pgm_sender_t::out_event (fd_t fd_)
{
// POLLOUT event from send socket. If write buffer is empty,
// try to read new data from the encoder.
if (write_size == 0) {
// First two bytes (sizeof uint16_t) are used to store message
// offset in following steps. Note that by passing our buffer to
// the get data function we prevent it from returning its own buffer.
unsigned char *bf = out_buffer + sizeof (uint16_t);
size_t bfsz = out_buffer_size - sizeof (uint16_t);
int offset = -1;
encoder.get_data (&bf, &bfsz, &offset);
// If there are no data to write stop polling for output.
if (!bfsz) {
reset_pollout (handle);
return;
}
// Put offset information in the buffer.
write_size = bfsz + sizeof (uint16_t);
put_uint16 (out_buffer, offset == -1 ? 0xffff : (uint16_t) offset);
}
if (has_tx_timer) {
cancel_timer (tx_timer_id);
has_tx_timer = false;
}
// Send the data.
size_t nbytes = pgm_socket.send (out_buffer, write_size);
// We can write either all data or 0 which means rate limit reached.
if (nbytes == write_size) {
write_size = 0;
} else {
xs_assert (nbytes == 0);
if (errno == ENOMEM) {
const long timeout = pgm_socket.get_tx_timeout ();
add_timer (timeout, tx_timer_id);
has_tx_timer = true;
} else
xs_assert (errno == EBUSY);
}
}
void xs::pgm_sender_t::timer_event (int token)
{
// Timer cancels on return by poller_base.
if (token == rx_timer_id) {
has_rx_timer = false;
in_event (retired_fd);
} else if (token == tx_timer_id) {
has_tx_timer = false;
out_event (retired_fd);
} else
xs_assert (false);
}
#endif
|
