aboutsummaryrefslogtreecommitdiff
path: root/src/transport/transport_api_monitor_plugins.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/transport/transport_api_monitor_plugins.c')
-rw-r--r--src/transport/transport_api_monitor_plugins.c285
1 files changed, 146 insertions, 139 deletions
diff --git a/src/transport/transport_api_monitor_plugins.c b/src/transport/transport_api_monitor_plugins.c
index 12d052008..7910a57c2 100644
--- a/src/transport/transport_api_monitor_plugins.c
+++ b/src/transport/transport_api_monitor_plugins.c
@@ -35,7 +35,8 @@
35/** 35/**
36 * Handle for a plugin session state monitor. 36 * Handle for a plugin session state monitor.
37 */ 37 */
38struct GNUNET_TRANSPORT_PluginMonitor { 38struct GNUNET_TRANSPORT_PluginMonitor
39{
39 /** 40 /**
40 * Connection to the service. 41 * Connection to the service.
41 */ 42 */
@@ -78,7 +79,8 @@ struct GNUNET_TRANSPORT_PluginMonitor {
78 * Abstract representation of a plugin's session. 79 * Abstract representation of a plugin's session.
79 * Corresponds to the `struct GNUNET_ATS_Session` within the TRANSPORT service. 80 * Corresponds to the `struct GNUNET_ATS_Session` within the TRANSPORT service.
80 */ 81 */
81struct GNUNET_TRANSPORT_PluginSession { 82struct GNUNET_TRANSPORT_PluginSession
83{
82 /** 84 /**
83 * Unique session identifier. 85 * Unique session identifier.
84 */ 86 */
@@ -98,7 +100,7 @@ struct GNUNET_TRANSPORT_PluginSession {
98 * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *` 100 * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *`
99 */ 101 */
100static void 102static void
101do_plugin_connect(void *cls); 103do_plugin_connect (void *cls);
102 104
103 105
104/** 106/**
@@ -110,23 +112,23 @@ do_plugin_connect(void *cls);
110 * @return #GNUNET_OK (continue to iterate) 112 * @return #GNUNET_OK (continue to iterate)
111 */ 113 */
112static int 114static int
113free_entry(void *cls, 115free_entry (void *cls,
114 uint32_t key, 116 uint32_t key,
115 void *value) 117 void *value)
116{ 118{
117 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls; 119 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
118 struct GNUNET_TRANSPORT_PluginSession *ps = value; 120 struct GNUNET_TRANSPORT_PluginSession *ps = value;
119 121
120 pm->cb(pm->cb_cls, 122 pm->cb (pm->cb_cls,
121 ps, 123 ps,
122 &ps->client_ctx, 124 &ps->client_ctx,
123 NULL); 125 NULL);
124 GNUNET_break(GNUNET_YES == 126 GNUNET_break (GNUNET_YES ==
125 GNUNET_CONTAINER_multihashmap32_remove(pm->sessions, 127 GNUNET_CONTAINER_multihashmap32_remove (pm->sessions,
126 key, 128 key,
127 ps)); 129 ps));
128 GNUNET_break(NULL == ps->client_ctx); 130 GNUNET_break (NULL == ps->client_ctx);
129 GNUNET_free(ps); 131 GNUNET_free (ps);
130 return GNUNET_OK; 132 return GNUNET_OK;
131} 133}
132 134
@@ -137,17 +139,17 @@ free_entry(void *cls,
137 * @param pm our context 139 * @param pm our context
138 */ 140 */
139static void 141static void
140reconnect_plugin_ctx(struct GNUNET_TRANSPORT_PluginMonitor *pm) 142reconnect_plugin_ctx (struct GNUNET_TRANSPORT_PluginMonitor *pm)
141{ 143{
142 GNUNET_MQ_destroy(pm->mq); 144 GNUNET_MQ_destroy (pm->mq);
143 pm->mq = NULL; 145 pm->mq = NULL;
144 GNUNET_CONTAINER_multihashmap32_iterate(pm->sessions, 146 GNUNET_CONTAINER_multihashmap32_iterate (pm->sessions,
145 &free_entry, 147 &free_entry,
146 pm); 148 pm);
147 pm->backoff = GNUNET_TIME_STD_BACKOFF(pm->backoff); 149 pm->backoff = GNUNET_TIME_STD_BACKOFF (pm->backoff);
148 pm->reconnect_task = GNUNET_SCHEDULER_add_delayed(pm->backoff, 150 pm->reconnect_task = GNUNET_SCHEDULER_add_delayed (pm->backoff,
149 &do_plugin_connect, 151 &do_plugin_connect,
150 pm); 152 pm);
151} 153}
152 154
153 155
@@ -158,16 +160,17 @@ reconnect_plugin_ctx(struct GNUNET_TRANSPORT_PluginMonitor *pm)
158 * @return 32-bit hash map index 160 * @return 32-bit hash map index
159 */ 161 */
160static uint32_t 162static uint32_t
161wrap_id(uint64_t id) 163wrap_id (uint64_t id)
162{ 164{
163 return ((uint32_t)id) ^ ((uint32_t)(id >> 32)); 165 return ((uint32_t) id) ^ ((uint32_t) (id >> 32));
164} 166}
165 167
166 168
167/** 169/**
168 * Context for #locate_by_id(). 170 * Context for #locate_by_id().
169 */ 171 */
170struct SearchContext { 172struct SearchContext
173{
171 /** 174 /**
172 * Result. 175 * Result.
173 */ 176 */
@@ -189,18 +192,18 @@ struct SearchContext {
189 * @return #GNUNET_OK (continue to iterate), or #GNUNET_SYSERR (match found) 192 * @return #GNUNET_OK (continue to iterate), or #GNUNET_SYSERR (match found)
190 */ 193 */
191static int 194static int
192locate_by_id(void *cls, 195locate_by_id (void *cls,
193 uint32_t key, 196 uint32_t key,
194 void *value) 197 void *value)
195{ 198{
196 struct SearchContext *sc = cls; 199 struct SearchContext *sc = cls;
197 struct GNUNET_TRANSPORT_PluginSession *ps = value; 200 struct GNUNET_TRANSPORT_PluginSession *ps = value;
198 201
199 if (sc->session_id == ps->session_id) 202 if (sc->session_id == ps->session_id)
200 { 203 {
201 sc->ps = ps; 204 sc->ps = ps;
202 return GNUNET_SYSERR; 205 return GNUNET_SYSERR;
203 } 206 }
204 return GNUNET_OK; 207 return GNUNET_OK;
205} 208}
206 209
@@ -213,25 +216,26 @@ locate_by_id(void *cls,
213 * @return #GNUNET_Ok if message is well-formed 216 * @return #GNUNET_Ok if message is well-formed
214 */ 217 */
215static int 218static int
216check_event(void *cls, 219check_event (void *cls,
217 const struct TransportPluginMonitorMessage *tpmm) 220 const struct TransportPluginMonitorMessage *tpmm)
218{ 221{
219 const char *pname; 222 const char *pname;
220 size_t pname_len; 223 size_t pname_len;
221 size_t paddr_len; 224 size_t paddr_len;
222 225
223 pname = (const char *)&tpmm[1]; 226 pname = (const char *) &tpmm[1];
224 pname_len = ntohs(tpmm->plugin_name_len); 227 pname_len = ntohs (tpmm->plugin_name_len);
225 paddr_len = ntohs(tpmm->plugin_address_len); 228 paddr_len = ntohs (tpmm->plugin_address_len);
226 if ((pname_len + 229 if ((pname_len
227 paddr_len + 230 + paddr_len
228 sizeof(struct TransportPluginMonitorMessage) != ntohs(tpmm->header.size)) || 231 + sizeof(struct TransportPluginMonitorMessage) != ntohs (
232 tpmm->header.size)) ||
229 ((0 != pname_len) && 233 ((0 != pname_len) &&
230 ('\0' != pname[pname_len - 1]))) 234 ('\0' != pname[pname_len - 1])))
231 { 235 {
232 GNUNET_break(0); 236 GNUNET_break (0);
233 return GNUNET_SYSERR; 237 return GNUNET_SYSERR;
234 } 238 }
235 return GNUNET_OK; 239 return GNUNET_OK;
236} 240}
237 241
@@ -243,8 +247,8 @@ check_event(void *cls,
243 * @paramm tpmm message with event data 247 * @paramm tpmm message with event data
244 */ 248 */
245static void 249static void
246handle_event(void *cls, 250handle_event (void *cls,
247 const struct TransportPluginMonitorMessage *tpmm) 251 const struct TransportPluginMonitorMessage *tpmm)
248{ 252{
249 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls; 253 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
250 struct GNUNET_TRANSPORT_PluginSession *ps; 254 struct GNUNET_TRANSPORT_PluginSession *ps;
@@ -257,63 +261,65 @@ handle_event(void *cls,
257 struct GNUNET_HELLO_Address addr; 261 struct GNUNET_HELLO_Address addr;
258 struct SearchContext rv; 262 struct SearchContext rv;
259 263
260 pname = (const char *)&tpmm[1]; 264 pname = (const char *) &tpmm[1];
261 pname_len = ntohs(tpmm->plugin_name_len); 265 pname_len = ntohs (tpmm->plugin_name_len);
262 paddr_len = ntohs(tpmm->plugin_address_len); 266 paddr_len = ntohs (tpmm->plugin_address_len);
263 paddr = &pname[pname_len]; 267 paddr = &pname[pname_len];
264 ps = NULL; 268 ps = NULL;
265 ss = (enum GNUNET_TRANSPORT_SessionState)ntohs(tpmm->session_state); 269 ss = (enum GNUNET_TRANSPORT_SessionState) ntohs (tpmm->session_state);
266 if (GNUNET_TRANSPORT_SS_INIT == ss) 270 if (GNUNET_TRANSPORT_SS_INIT == ss)
267 { 271 {
268 ps = GNUNET_new(struct GNUNET_TRANSPORT_PluginSession); 272 ps = GNUNET_new (struct GNUNET_TRANSPORT_PluginSession);
269 ps->session_id = tpmm->session_id; 273 ps->session_id = tpmm->session_id;
270 (void)GNUNET_CONTAINER_multihashmap32_put(pm->sessions, 274 (void) GNUNET_CONTAINER_multihashmap32_put (pm->sessions,
271 wrap_id(tpmm->session_id), 275 wrap_id (tpmm->session_id),
272 ps, 276 ps,
273 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE); 277 GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
274 } 278 }
275 else 279 else
276 { 280 {
277 rv.session_id = tpmm->session_id; 281 rv.session_id = tpmm->session_id;
278 rv.ps = NULL; 282 rv.ps = NULL;
279 (void)GNUNET_CONTAINER_multihashmap32_get_multiple(pm->sessions, 283 (void) GNUNET_CONTAINER_multihashmap32_get_multiple (pm->sessions,
280 wrap_id(tpmm->session_id), 284 wrap_id (
285 tpmm->session_id),
281 &locate_by_id, 286 &locate_by_id,
282 &rv); 287 &rv);
283 ps = rv.ps; 288 ps = rv.ps;
284 if (NULL == ps) 289 if (NULL == ps)
285 { 290 {
286 GNUNET_break(0); 291 GNUNET_break (0);
287 reconnect_plugin_ctx(pm); 292 reconnect_plugin_ctx (pm);
288 return; 293 return;
289 }
290 } 294 }
295 }
291 info.state = ss; 296 info.state = ss;
292 info.is_inbound = (int16_t)ntohs(tpmm->is_inbound); 297 info.is_inbound = (int16_t) ntohs (tpmm->is_inbound);
293 info.num_msg_pending = ntohl(tpmm->msgs_pending); 298 info.num_msg_pending = ntohl (tpmm->msgs_pending);
294 info.num_bytes_pending = ntohl(tpmm->bytes_pending); 299 info.num_bytes_pending = ntohl (tpmm->bytes_pending);
295 info.receive_delay = GNUNET_TIME_absolute_ntoh(tpmm->delay); 300 info.receive_delay = GNUNET_TIME_absolute_ntoh (tpmm->delay);
296 info.session_timeout = GNUNET_TIME_absolute_ntoh(tpmm->timeout); 301 info.session_timeout = GNUNET_TIME_absolute_ntoh (tpmm->timeout);
297 info.address = &addr; 302 info.address = &addr;
298 addr.peer = tpmm->peer; 303 addr.peer = tpmm->peer;
299 addr.address = (0 == paddr_len) ? NULL : paddr; 304 addr.address = (0 == paddr_len) ? NULL : paddr;
300 addr.address_length = paddr_len; 305 addr.address_length = paddr_len;
301 addr.transport_name = (0 == pname_len) ? NULL : pname; 306 addr.transport_name = (0 == pname_len) ? NULL : pname;
302 addr.local_info = GNUNET_HELLO_ADDRESS_INFO_NONE; 307 addr.local_info = GNUNET_HELLO_ADDRESS_INFO_NONE;
303 pm->cb(pm->cb_cls, 308 pm->cb (pm->cb_cls,
304 ps, 309 ps,
305 &ps->client_ctx, 310 &ps->client_ctx,
306 &info); 311 &info);
307 312
308 if (GNUNET_TRANSPORT_SS_DONE == ss) 313 if (GNUNET_TRANSPORT_SS_DONE == ss)
309 { 314 {
310 GNUNET_break(NULL == ps->client_ctx); 315 GNUNET_break (NULL == ps->client_ctx);
311 GNUNET_assert(GNUNET_YES == 316 GNUNET_assert (GNUNET_YES ==
312 GNUNET_CONTAINER_multihashmap32_remove(pm->sessions, 317 GNUNET_CONTAINER_multihashmap32_remove (pm->sessions,
313 wrap_id(tpmm->session_id), 318 wrap_id (
319 tpmm->session_id),
314 ps)); 320 ps));
315 GNUNET_free(ps); 321 GNUNET_free (ps);
316 } 322 }
317} 323}
318 324
319 325
@@ -324,16 +330,16 @@ handle_event(void *cls,
324 * @param msg message from the service 330 * @param msg message from the service
325 */ 331 */
326static void 332static void
327handle_sync(void *cls, 333handle_sync (void *cls,
328 const struct GNUNET_MessageHeader *msg) 334 const struct GNUNET_MessageHeader *msg)
329{ 335{
330 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls; 336 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
331 337
332 /* we are in sync, notify callback */ 338 /* we are in sync, notify callback */
333 pm->cb(pm->cb_cls, 339 pm->cb (pm->cb_cls,
334 NULL, 340 NULL,
335 NULL, 341 NULL,
336 NULL); 342 NULL);
337} 343}
338 344
339 345
@@ -347,12 +353,12 @@ handle_sync(void *cls,
347 * @param error error code 353 * @param error error code
348 */ 354 */
349static void 355static void
350mq_error_handler(void *cls, 356mq_error_handler (void *cls,
351 enum GNUNET_MQ_Error error) 357 enum GNUNET_MQ_Error error)
352{ 358{
353 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls; 359 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
354 360
355 reconnect_plugin_ctx(pm); 361 reconnect_plugin_ctx (pm);
356} 362}
357 363
358 364
@@ -362,35 +368,35 @@ mq_error_handler(void *cls,
362 * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *` 368 * @param cls our `struct GNUNET_TRANSPORT_PluginMonitor *`
363 */ 369 */
364static void 370static void
365do_plugin_connect(void *cls) 371do_plugin_connect (void *cls)
366{ 372{
367 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls; 373 struct GNUNET_TRANSPORT_PluginMonitor *pm = cls;
368 struct GNUNET_MQ_MessageHandler handlers[] = { 374 struct GNUNET_MQ_MessageHandler handlers[] = {
369 GNUNET_MQ_hd_var_size(event, 375 GNUNET_MQ_hd_var_size (event,
370 GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_EVENT, 376 GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_EVENT,
371 struct TransportPluginMonitorMessage, 377 struct TransportPluginMonitorMessage,
372 pm), 378 pm),
373 GNUNET_MQ_hd_fixed_size(sync, 379 GNUNET_MQ_hd_fixed_size (sync,
374 GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_SYNC, 380 GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_SYNC,
375 struct GNUNET_MessageHeader, 381 struct GNUNET_MessageHeader,
376 pm), 382 pm),
377 GNUNET_MQ_handler_end() 383 GNUNET_MQ_handler_end ()
378 }; 384 };
379 struct GNUNET_MessageHeader *msg; 385 struct GNUNET_MessageHeader *msg;
380 struct GNUNET_MQ_Envelope *env; 386 struct GNUNET_MQ_Envelope *env;
381 387
382 pm->reconnect_task = NULL; 388 pm->reconnect_task = NULL;
383 pm->mq = GNUNET_CLIENT_connect(pm->cfg, 389 pm->mq = GNUNET_CLIENT_connect (pm->cfg,
384 "transport", 390 "transport",
385 handlers, 391 handlers,
386 &mq_error_handler, 392 &mq_error_handler,
387 pm); 393 pm);
388 if (NULL == pm->mq) 394 if (NULL == pm->mq)
389 return; 395 return;
390 env = GNUNET_MQ_msg(msg, 396 env = GNUNET_MQ_msg (msg,
391 GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_START); 397 GNUNET_MESSAGE_TYPE_TRANSPORT_MONITOR_PLUGIN_START);
392 GNUNET_MQ_send(pm->mq, 398 GNUNET_MQ_send (pm->mq,
393 env); 399 env);
394} 400}
395 401
396 402
@@ -404,23 +410,23 @@ do_plugin_connect(void *cls)
404 * @return NULL on error, otherwise handle for cancellation 410 * @return NULL on error, otherwise handle for cancellation
405 */ 411 */
406struct GNUNET_TRANSPORT_PluginMonitor * 412struct GNUNET_TRANSPORT_PluginMonitor *
407GNUNET_TRANSPORT_monitor_plugins(const struct GNUNET_CONFIGURATION_Handle *cfg, 413GNUNET_TRANSPORT_monitor_plugins (const struct GNUNET_CONFIGURATION_Handle *cfg,
408 GNUNET_TRANSPORT_SessionMonitorCallback cb, 414 GNUNET_TRANSPORT_SessionMonitorCallback cb,
409 void *cb_cls) 415 void *cb_cls)
410{ 416{
411 struct GNUNET_TRANSPORT_PluginMonitor *pm; 417 struct GNUNET_TRANSPORT_PluginMonitor *pm;
412 418
413 pm = GNUNET_new(struct GNUNET_TRANSPORT_PluginMonitor); 419 pm = GNUNET_new (struct GNUNET_TRANSPORT_PluginMonitor);
414 pm->cb = cb; 420 pm->cb = cb;
415 pm->cb_cls = cb_cls; 421 pm->cb_cls = cb_cls;
416 pm->cfg = cfg; 422 pm->cfg = cfg;
417 do_plugin_connect(pm); 423 do_plugin_connect (pm);
418 if (NULL == pm->mq) 424 if (NULL == pm->mq)
419 { 425 {
420 GNUNET_free(pm); 426 GNUNET_free (pm);
421 return NULL; 427 return NULL;
422 } 428 }
423 pm->sessions = GNUNET_CONTAINER_multihashmap32_create(128); 429 pm->sessions = GNUNET_CONTAINER_multihashmap32_create (128);
424 return pm; 430 return pm;
425} 431}
426 432
@@ -434,23 +440,24 @@ GNUNET_TRANSPORT_monitor_plugins(const struct GNUNET_CONFIGURATION_Handle *cfg,
434 * @param pm handle of the request that is to be cancelled 440 * @param pm handle of the request that is to be cancelled
435 */ 441 */
436void 442void
437GNUNET_TRANSPORT_monitor_plugins_cancel(struct GNUNET_TRANSPORT_PluginMonitor *pm) 443GNUNET_TRANSPORT_monitor_plugins_cancel (struct
444 GNUNET_TRANSPORT_PluginMonitor *pm)
438{ 445{
439 if (NULL != pm->mq) 446 if (NULL != pm->mq)
440 { 447 {
441 GNUNET_MQ_destroy(pm->mq); 448 GNUNET_MQ_destroy (pm->mq);
442 pm->mq = NULL; 449 pm->mq = NULL;
443 } 450 }
444 if (NULL != pm->reconnect_task) 451 if (NULL != pm->reconnect_task)
445 { 452 {
446 GNUNET_SCHEDULER_cancel(pm->reconnect_task); 453 GNUNET_SCHEDULER_cancel (pm->reconnect_task);
447 pm->reconnect_task = NULL; 454 pm->reconnect_task = NULL;
448 } 455 }
449 GNUNET_CONTAINER_multihashmap32_iterate(pm->sessions, 456 GNUNET_CONTAINER_multihashmap32_iterate (pm->sessions,
450 &free_entry, 457 &free_entry,
451 pm); 458 pm);
452 GNUNET_CONTAINER_multihashmap32_destroy(pm->sessions); 459 GNUNET_CONTAINER_multihashmap32_destroy (pm->sessions);
453 GNUNET_free(pm); 460 GNUNET_free (pm);
454} 461}
455 462
456 463