GNUnet  0.10.x
fragmentation.c
Go to the documentation of this file.
1 /*
2  This file is part of GNUnet
3  Copyright (C) 2009-2013 GNUnet e.V.
4 
5  GNUnet is free software: you can redistribute it and/or modify it
6  under the terms of the GNU Affero General Public License as published
7  by the Free Software Foundation, either version 3 of the License,
8  or (at your option) any later version.
9 
10  GNUnet is distributed in the hope that it will be useful, but
11  WITHOUT ANY WARRANTY; without even the implied warranty of
12  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13  Affero General Public License for more details.
14 
15  You should have received a copy of the GNU Affero General Public License
16  along with this program. If not, see <http://www.gnu.org/licenses/>.
17 
18  SPDX-License-Identifier: AGPL3.0-or-later
19 */
25 #include "platform.h"
27 #include "gnunet_protocols.h"
28 #include "fragmentation.h"
29 
30 
34 #define MIN_ACK_DELAY GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_MILLISECONDS, 1)
35 
36 
41 {
46 
51 
56 
61 
66 
71 
75  const struct GNUNET_MessageHeader *msg;
76 
81 
85  void *proc_cls;
86 
90  uint64_t acks;
91 
96  uint64_t acks_mask;
97 
102 
106  uint32_t fragment_id;
107 
111  unsigned int next_transmission;
112 
116  unsigned int num_rounds;
117 
121  unsigned int num_transmissions;
122 
126  int8_t proc_busy;
127 
131  int8_t wack;
132 
136  uint16_t mtu;
137 
138 };
139 
140 
147 const char *
149 {
150  static char buf[128];
151  const struct FragmentAcknowledgement *fa;
152 
153  if (sizeof (struct FragmentAcknowledgement) !=
154  htons (ack->size))
155  return "<malformed ack>";
156  fa = (const struct FragmentAcknowledgement *) ack;
157  GNUNET_snprintf (buf,
158  sizeof (buf),
159  "%u-%llX",
160  ntohl (fa->fragment_id),
161  GNUNET_ntohll (fa->bits));
162  return buf;
163 }
164 
165 
171 static void
172 transmit_next (void *cls)
173 {
174  struct GNUNET_FRAGMENT_Context *fc = cls;
175  char msg[fc->mtu];
176  const char *mbuf;
177  struct FragmentHeader *fh;
178  struct GNUNET_TIME_Relative delay;
179  unsigned int bit;
180  size_t size;
181  size_t fsize;
182  int wrap;
183 
184  fc->task = NULL;
186  if (0 == fc->acks)
187  return; /* all done */
188  /* calculate delay */
189  wrap = 0;
190  while (0 == (fc->acks & (1LLU << fc->next_transmission)))
191  {
192  fc->next_transmission = (fc->next_transmission + 1) % 64;
193  wrap |= (0 == fc->next_transmission);
194  }
195  bit = fc->next_transmission;
196  size = ntohs (fc->msg->size);
197  if (bit == size / (fc->mtu - sizeof (struct FragmentHeader)))
198  fsize =
199  (size % (fc->mtu - sizeof (struct FragmentHeader))) +
200  sizeof (struct FragmentHeader);
201  else
202  fsize = fc->mtu;
203  if (NULL != fc->tracker)
205  fsize);
206  else
207  delay = GNUNET_TIME_UNIT_ZERO;
208  if (delay.rel_value_us > 0)
209  {
211  "Fragmentation logic delays transmission of next fragment by %s\n",
213  GNUNET_YES));
214  fc->task = GNUNET_SCHEDULER_add_delayed (delay,
215  &transmit_next,
216  fc);
217  return;
218  }
219  fc->next_transmission = (fc->next_transmission + 1) % 64;
220  wrap |= (0 == fc->next_transmission);
221  while (0 == (fc->acks & (1LLU << fc->next_transmission)))
222  {
223  fc->next_transmission = (fc->next_transmission + 1) % 64;
224  wrap |= (0 == fc->next_transmission);
225  }
226 
227  /* assemble fragmentation message */
228  mbuf = (const char *) &fc[1];
229  fh = (struct FragmentHeader *) msg;
230  fh->header.size = htons (fsize);
232  fh->fragment_id = htonl (fc->fragment_id);
233  fh->total_size = fc->msg->size; /* already in big-endian */
234  fh->offset = htons ((fc->mtu - sizeof (struct FragmentHeader)) * bit);
235  GNUNET_memcpy (&fh[1], &mbuf[bit * (fc->mtu - sizeof (struct FragmentHeader))],
236  fsize - sizeof (struct FragmentHeader));
237  if (NULL != fc->tracker)
240  _("# fragments transmitted"),
241  1,
242  GNUNET_NO);
243  if (0 != fc->last_round.abs_value_us)
245  _("# fragments retransmitted"),
246  1,
247  GNUNET_NO);
248 
249  /* select next message to calculate delay */
250  bit = fc->next_transmission;
251  size = ntohs (fc->msg->size);
252  if (bit == size / (fc->mtu - sizeof (struct FragmentHeader)))
253  fsize = size % (fc->mtu - sizeof (struct FragmentHeader));
254  else
255  fsize = fc->mtu;
256  if (NULL != fc->tracker)
258  fsize);
259  else
260  delay = GNUNET_TIME_UNIT_ZERO;
261  if (fc->num_rounds < 64)
262  delay = GNUNET_TIME_relative_max (delay,
264  (fc->msg_delay,
265  (1ULL << fc->num_rounds)));
266  else
268  if (wrap)
269  {
270  /* full round transmitted wait 2x delay for ACK before going again */
271  fc->num_rounds++;
273  /* never use zero, need some time for ACK always */
274  delay = GNUNET_TIME_relative_max (MIN_ACK_DELAY, delay);
275  fc->wack = GNUNET_YES;
278  _("# fragments wrap arounds"),
279  1,
280  GNUNET_NO);
281  }
282  fc->proc_busy = GNUNET_YES;
284  fc->num_transmissions++;
285  fc->proc (fc->proc_cls,
286  &fh->header);
287 }
288 
289 
312  uint16_t mtu,
316  const struct GNUNET_MessageHeader *msg,
318  void *proc_cls)
319 {
320  struct GNUNET_FRAGMENT_Context *fc;
321  size_t size;
322  uint64_t bits;
323 
325  _("# messages fragmented"),
326  1,
327  GNUNET_NO);
328  GNUNET_assert (mtu >= 1024 + sizeof (struct FragmentHeader));
329  size = ntohs (msg->size);
331  _("# total size of fragmented messages"),
332  size, GNUNET_NO);
333  GNUNET_assert (size >= sizeof (struct GNUNET_MessageHeader));
334  fc = GNUNET_malloc (sizeof (struct GNUNET_FRAGMENT_Context) + size);
335  fc->stats = stats;
336  fc->mtu = mtu;
337  fc->tracker = tracker;
338  fc->ack_delay = ack_delay;
339  fc->msg_delay = msg_delay;
340  fc->msg = (const struct GNUNET_MessageHeader *) &fc[1];
341  fc->proc = proc;
342  fc->proc_cls = proc_cls;
343  fc->fragment_id =
345  UINT32_MAX);
346  GNUNET_memcpy (&fc[1], msg, size);
347  bits =
348  (size + mtu - sizeof (struct FragmentHeader) - 1) / (mtu -
349  sizeof (struct
350  FragmentHeader));
351  GNUNET_assert (bits <= 64);
352  if (bits == 64)
353  fc->acks_mask = UINT64_MAX; /* set all 64 bit */
354  else
355  fc->acks_mask = (1LLU << bits) - 1; /* set lowest 'bits' bit */
356  fc->acks = fc->acks_mask;
358  return fc;
359 }
360 
361 
369 void
371 {
373  fc->proc_busy = GNUNET_NO;
374  GNUNET_assert (fc->task == NULL);
375  fc->task =
377  &transmit_next,
378  fc);
379 }
380 
381 
393 int
395  const struct GNUNET_MessageHeader *msg)
396 {
397  const struct FragmentAcknowledgement *fa;
398  uint64_t abits;
399  struct GNUNET_TIME_Relative ndelay;
400  unsigned int ack_cnt;
401  unsigned int snd_cnt;
402  unsigned int i;
403 
404  if (sizeof (struct FragmentAcknowledgement) != ntohs (msg->size))
405  {
406  GNUNET_break_op (0);
407  return GNUNET_SYSERR;
408  }
409  fa = (const struct FragmentAcknowledgement *) msg;
410  if (ntohl (fa->fragment_id) != fc->fragment_id)
411  return GNUNET_SYSERR; /* not our ACK */
412  abits = GNUNET_ntohll (fa->bits);
413  if ( (GNUNET_YES == fc->wack) &&
414  (0 != fc->num_transmissions) )
415  {
416  /* normal ACK, can update running average of delay... */
417  fc->wack = GNUNET_NO;
419  fc->ack_delay.rel_value_us =
420  (ndelay.rel_value_us / fc->num_transmissions + 3 * fc->ack_delay.rel_value_us) / 4;
421  /* calculate ratio msg sent vs. msg acked */
422  ack_cnt = 0;
423  snd_cnt = 0;
424  for (i=0;i<64;i++)
425  {
426  if (1 == (fc->acks_mask & (1ULL << i)))
427  {
428  snd_cnt++;
429  if (0 == (abits & (1ULL << i)))
430  ack_cnt++;
431  }
432  }
433  if (0 == ack_cnt)
434  {
435  /* complete loss */
437  snd_cnt);
438  }
439  else if (snd_cnt > ack_cnt)
440  {
441  /* some loss, slow down proportionally */
442  fc->msg_delay.rel_value_us = ((fc->msg_delay.rel_value_us * ack_cnt) / snd_cnt);
443  }
444  else if (snd_cnt == ack_cnt)
445  {
446  fc->msg_delay.rel_value_us =
447  (ndelay.rel_value_us / fc->num_transmissions + 3 * fc->msg_delay.rel_value_us) / 5;
448  }
449  fc->num_transmissions = 0;
454  }
456  _("# fragment acknowledgements received"),
457  1,
458  GNUNET_NO);
459  if (abits != (fc->acks & abits))
460  {
461  /* ID collission or message reordering, count! This should be rare! */
463  _("# bits removed from fragmentation ACKs"), 1,
464  GNUNET_NO);
465  }
466  fc->acks = abits & fc->acks_mask;
467  if (0 != fc->acks)
468  {
469  /* more to transmit, do so right now (if tracker permits...) */
470  if (fc->task != NULL)
471  {
472  /* schedule next transmission now, no point in waiting... */
475  }
476  else
477  {
478  /* only case where there is no task should be if we're waiting
479  * for the right to transmit again (proc_busy set to YES) */
481  }
482  return GNUNET_NO;
483  }
484 
485  /* all done */
487  _("# fragmentation transmissions completed"),
488  1,
489  GNUNET_NO);
490  if (NULL != fc->task)
491  {
493  fc->task = NULL;
494  }
495  return GNUNET_OK;
496 }
497 
498 
509 void
513 {
514  if (fc->task != NULL)
516  if (NULL != ack_delay)
517  *ack_delay = fc->ack_delay;
518  if (NULL != msg_delay)
520  fc->num_rounds);
521  GNUNET_free (fc);
522 }
523 
524 
525 /* end of fragmentation.c */
unsigned int num_rounds
How many rounds of transmission have we completed so far?
uint32_t fragment_id
Unique fragment ID.
Definition: fragmentation.h:77
uint64_t acks
Bitfield, set to 1 for each unacknowledged fragment.
Definition: fragmentation.c:90
struct GNUNET_MessageHeader header
Message header.
Definition: fragmentation.h:42
int GNUNET_BANDWIDTH_tracker_consume(struct GNUNET_BANDWIDTH_Tracker *av, ssize_t size)
Notify the tracker that a certain number of bytes of bandwidth have been consumed.
Definition: bandwidth.c:402
uint32_t fragment_id
Unique fragment ID.
Definition: fragmentation.h:47
struct GNUNET_SCHEDULER_Task * task
Task performing work for the fragmenter.
uint64_t rel_value_us
The actual value.
struct GNUNET_TIME_Absolute last_round
Time we transmitted the last message of the last round.
Definition: fragmentation.c:70
int GNUNET_snprintf(char *buf, size_t size, const char *format,...)
Like snprintf, just aborts if the buffer is of insufficient size.
uint64_t bits
Bits that are being acknowledged, in big-endian.
Definition: fragmentation.h:84
struct GNUNET_TIME_Relative GNUNET_TIME_relative_max(struct GNUNET_TIME_Relative t1, struct GNUNET_TIME_Relative t2)
Return the maximum of two relative time values.
Definition: time.c:286
struct GNUNET_STATISTICS_Handle * stats
Statistics to use.
Definition: fragmentation.c:45
struct GNUNET_TIME_Absolute GNUNET_TIME_relative_to_absolute(struct GNUNET_TIME_Relative rel)
Convert relative time to an absolute time in the future.
Definition: time.c:245
uint32_t GNUNET_CRYPTO_random_u32(enum GNUNET_CRYPTO_Quality mode, uint32_t i)
Produce a random value.
#define GNUNET_assert(cond)
Use this for fatal errors that cannot be handled.
#define GNUNET_TIME_UNIT_SECONDS
One second.
library to help fragment messages
struct GNUNET_TIME_Relative msg_delay
Current expected delay between messages.
Definition: fragmentation.c:60
void(* GNUNET_FRAGMENT_MessageProcessor)(void *cls, const struct GNUNET_MessageHeader *msg)
Function that is called with messages created by the fragmentation module.
GNUNET_FRAGMENT_MessageProcessor proc
Function to call for transmissions.
Definition: fragmentation.c:80
struct GNUNET_FRAGMENT_Context * GNUNET_FRAGMENT_context_create(struct GNUNET_STATISTICS_Handle *stats, uint16_t mtu, struct GNUNET_BANDWIDTH_Tracker *tracker, struct GNUNET_TIME_Relative msg_delay, struct GNUNET_TIME_Relative ack_delay, const struct GNUNET_MessageHeader *msg, GNUNET_FRAGMENT_MessageProcessor proc, void *proc_cls)
Create a fragmentation context for the given message.
uint64_t acks_mask
Bitfield with all possible bits for acks (used to mask the ack we get back).
Definition: fragmentation.c:96
Struct to track available bandwidth.
uint16_t total_size
Total message size of the original message.
Definition: fragmentation.h:52
#define GNUNET_NO
Definition: gnunet_common.h:81
#define GNUNET_OK
Named constants for return values.
Definition: gnunet_common.h:78
uint16_t size
The length of the struct (in bytes, including the length field itself), in big-endian format...
void GNUNET_STATISTICS_update(struct GNUNET_STATISTICS_Handle *handle, const char *name, int64_t delta, int make_persistent)
Set statistic value for the peer.
Handle for the service.
uint64_t abs_value_us
The actual value.
int8_t proc_busy
GNUNET_YES if we called proc and are now waiting for GNUNET_FRAGMENT_context_transmission_done() ...
const char * GNUNET_FRAGMENT_print_ack(const struct GNUNET_MessageHeader *ack)
Convert an ACK message to a printable format suitable for logging.
struct GNUNET_BANDWIDTH_Tracker * tracker
Tracker for flow control.
Definition: fragmentation.c:50
void * proc_cls
Closure for proc.
Definition: fragmentation.c:85
void GNUNET_FRAGMENT_context_transmission_done(struct GNUNET_FRAGMENT_Context *fc)
Continuation to call from the &#39;proc&#39; function after the fragment has been transmitted (and hence the ...
#define _(String)
GNU gettext support macro.
Definition: platform.h:208
uint16_t mtu
Target fragment size.
uint32_t fragment_id
Our fragmentation ID.
Fragmentation context.
Definition: fragmentation.c:40
struct GNUNET_SCHEDULER_Task * GNUNET_SCHEDULER_add_delayed(struct GNUNET_TIME_Relative delay, GNUNET_SCHEDULER_TaskCallback task, void *task_cls)
Schedule a new task to be run with a specified delay.
Definition: scheduler.c:1246
uint16_t type
The type of the message (GNUNET_MESSAGE_TYPE_XXXX), in big-endian format.
#define GNUNET_memcpy(dst, src, n)
unsigned int num_transmissions
How many transmission have we completed in this round?
#define GNUNET_break_op(cond)
Use this for assertion violations caused by other peers (i.e.
struct GNUNET_SCHEDULER_Task * GNUNET_SCHEDULER_add_now(GNUNET_SCHEDULER_TaskCallback task, void *task_cls)
Schedule a new task to be run as soon as possible.
Definition: scheduler.c:1273
const char * GNUNET_STRINGS_relative_time_to_string(struct GNUNET_TIME_Relative delta, int do_round)
Give relative time in human-readable fancy format.
Definition: strings.c:727
static void transmit_next(void *cls)
Transmit the next fragment to the other peer.
static char buf[2048]
#define GNUNET_TIME_UNIT_FOREVER_REL
Constant used to specify "forever".
static int fh
Handle to the unique file.
const struct GNUNET_MessageHeader * msg
Message to fragment (allocated at the end of this struct).
Definition: fragmentation.c:75
struct GNUNET_TIME_Absolute GNUNET_TIME_absolute_get(void)
Get the current time.
Definition: time.c:118
int GNUNET_FRAGMENT_process_ack(struct GNUNET_FRAGMENT_Context *fc, const struct GNUNET_MessageHeader *msg)
Process an acknowledgement message we got from the other side (to control re-transmits).
struct GNUNET_TIME_Relative GNUNET_BANDWIDTH_tracker_get_delay(struct GNUNET_BANDWIDTH_Tracker *av, size_t size)
Compute how long we should wait until consuming size bytes of bandwidth in order to stay within the g...
Definition: bandwidth.c:458
struct GNUNET_TIME_Relative GNUNET_TIME_relative_min(struct GNUNET_TIME_Relative t1, struct GNUNET_TIME_Relative t2)
Return the minimum of two relative time values.
Definition: time.c:271
#define GNUNET_SYSERR
Definition: gnunet_common.h:79
static unsigned int size
Size of the "table".
Definition: peer.c:67
uint16_t offset
Absolute offset (in bytes) of this fragment in the original message.
Definition: fragmentation.h:58
#define GNUNET_TIME_UNIT_ZERO
Relative time zero.
int8_t wack
GNUNET_YES if we are waiting for an ACK.
struct GNUNET_TIME_Relative GNUNET_TIME_absolute_get_duration(struct GNUNET_TIME_Absolute whence)
Get the duration of an operation as the difference of the current time and the given start time "henc...
Definition: time.c:373
#define GNUNET_log(kind,...)
Message fragment acknowledgement.
Definition: fragmentation.h:66
Entry in list of pending tasks.
Definition: scheduler.c:134
Header for all communications.
Time for absolute times used by GNUnet, in microseconds.
struct GNUNET_TIME_Relative ack_delay
Current expected delay for ACKs.
Definition: fragmentation.c:55
#define GNUNET_YES
Definition: gnunet_common.h:80
#define MIN_ACK_DELAY
Absolute minimum delay we impose between sending and expecting ACK to arrive.
Definition: fragmentation.c:34
unsigned int next_transmission
Round-robin selector for the next transmission.
#define GNUNET_MESSAGE_TYPE_FRAGMENT
FRAGMENT of a larger message.
struct GNUNET_SCHEDULER_Task * GNUNET_SCHEDULER_add_at(struct GNUNET_TIME_Absolute at, GNUNET_SCHEDULER_TaskCallback task, void *task_cls)
Schedule a new task to be run at the specified time.
Definition: scheduler.c:1223
void GNUNET_FRAGMENT_context_destroy(struct GNUNET_FRAGMENT_Context *fc, struct GNUNET_TIME_Relative *msg_delay, struct GNUNET_TIME_Relative *ack_delay)
Destroy the given fragmentation context (stop calling &#39;proc&#39;, free resources).
struct GNUNET_TIME_Relative GNUNET_TIME_relative_saturating_multiply(struct GNUNET_TIME_Relative rel, unsigned long long factor)
Saturating multiply relative time by a given factor.
Definition: time.c:499
struct GNUNET_TIME_Absolute delay_until
Next allowed transmission time.
Definition: fragmentation.c:65
No good quality of the operation is needed (i.e., random numbers can be pseudo-random).
#define GNUNET_malloc(size)
Wrapper around malloc.
uint64_t GNUNET_ntohll(uint64_t n)
Convert unsigned 64-bit integer to host byte order.
Definition: common_endian.c:48
#define GNUNET_free(ptr)
Wrapper around free.
Time for relative time used by GNUnet, in microseconds.
Header for a message fragment.
Definition: fragmentation.h:36
void * GNUNET_SCHEDULER_cancel(struct GNUNET_SCHEDULER_Task *task)
Cancel the task with the specified identifier.
Definition: scheduler.c:965