src/network.c

   1 /**
   2  * collectd - src/network.c
   3  * Copyright (C) 2005-2007  Florian octo Forster
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; only version 2 of the License is applicable.
   8  *
   9  * This program is distributed in the hope that it will be useful, but
  10  * WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU General Public License along
  15  * with this program; if not, write to the Free Software Foundation, Inc.,
  16  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  17  *
  18  * Authors:
  19  *   Florian octo Forster <octo at verplant.org>
  20  **/
  21
  22 #include "collectd.h"
  23 #include "plugin.h"
  24 #include "common.h"
  25 #include "configfile.h"
  26 #include "utils_avltree.h"
  27
  28 #include "network.h"
  29
  30 #if HAVE_PTHREAD_H
  31 # include <pthread.h>
  32 #endif
  33 #if HAVE_SYS_SOCKET_H
  34 # include <sys/socket.h>
  35 #endif
  36 #if HAVE_NETDB_H
  37 # include <netdb.h>
  38 #endif
  39 #if HAVE_NETINET_IN_H
  40 # include <netinet/in.h>
  41 #endif
  42 #if HAVE_ARPA_INET_H
  43 # include <arpa/inet.h>
  44 #endif
  45 #if HAVE_POLL_H
  46 # include <poll.h>
  47 #endif
  48
  49 /* 1500 - 40 - 8  =  Ethernet packet - IPv6 header - UDP header */
  50 /* #define BUFF_SIZE 1452 */
  51
  52 #ifndef IPV6_ADD_MEMBERSHIP
  53 # ifdef IPV6_JOIN_GROUP
  54 #  define IPV6_ADD_MEMBERSHIP IPV6_JOIN_GROUP
  55 # else
  56 #  error "Neither IP_ADD_MEMBERSHIP nor IPV6_JOIN_GROUP is defined"
  57 # endif
  58 #endif /* !IP_ADD_MEMBERSHIP */
  59
  60 #define BUFF_SIZE 1024
  61
  62 /*
  63  * Private data types
  64  */
  65 typedef struct sockent
  66 {
  67         int                      fd;
  68         struct sockaddr_storage *addr;
  69         socklen_t                addrlen;
  70         struct sockent          *next;
  71 } sockent_t;
  72
  73 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
  74  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  75  * +-------+-----------------------+-------------------------------+
  76  * ! Ver.  !                       ! Length                        !
  77  * +-------+-----------------------+-------------------------------+
  78  */
  79 struct part_header_s
  80 {
  81         uint16_t type;
  82         uint16_t length;
  83 };
  84 typedef struct part_header_s part_header_t;
  85
  86 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
  87  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
  88  * +-------------------------------+-------------------------------+
  89  * ! Type                          ! Length                        !
  90  * +-------------------------------+-------------------------------+
  91  * : (Length - 4) Bytes                                            :
  92  * +---------------------------------------------------------------+
  93  */
  94 struct part_string_s
  95 {
  96         part_header_t *head;
  97         char *value;
  98 };
  99 typedef struct part_string_s part_string_t;
 100
 101 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
 102  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 103  * +-------------------------------+-------------------------------+
 104  * ! Type                          ! Length                        !
 105  * +-------------------------------+-------------------------------+
 106  * : (Length - 4 == 2 || 4 || 8) Bytes                             :
 107  * +---------------------------------------------------------------+
 108  */
 109 struct part_number_s
 110 {
 111         part_header_t *head;
 112         uint64_t *value;
 113 };
 114 typedef struct part_number_s part_number_t;
 115
 116 /*                      1 1 1 1 1 1 1 1 1 1 2 2 2 2 2 2 2 2 2 2 3 3
 117  *  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 118  * +-------------------------------+-------------------------------+
 119  * ! Type                          ! Length                        !
 120  * +-------------------------------+---------------+---------------+
 121  * ! Num of values                 ! Type0         ! Type1         !
 122  * +-------------------------------+---------------+---------------+
 123  * ! Value0                                                        !
 124  * !                                                               !
 125  * +---------------------------------------------------------------+
 126  * ! Value1                                                        !
 127  * !                                                               !
 128  * +---------------------------------------------------------------+
 129  */
 130 struct part_values_s
 131 {
 132         part_header_t *head;
 133         uint16_t *num_values;
 134         uint8_t  *values_types;
 135         value_t  *values;
 136 };
 137 typedef struct part_values_s part_values_t;
 138
 139 struct receive_list_entry_s
 140 {
 141   char data[BUFF_SIZE];
 142   int  data_len;
 143   struct receive_list_entry_s *next;
 144 };
 145 typedef struct receive_list_entry_s receive_list_entry_t;
 146
 147 /*
 148  * Private variables
 149  */
 150 static const char *config_keys[] =
 151 {
 152         "CacheFlush",
 153         "Listen",
 154         "Server",
 155         "TimeToLive",
 156         "Forward"
 157 };
 158 static int config_keys_num = STATIC_ARRAY_SIZE (config_keys);
 159
 160 static int network_config_ttl = 0;
 161 static int network_config_forward = 0;
 162
 163 static sockent_t *sending_sockets = NULL;
 164
 165 static receive_list_entry_t *receive_list_head = NULL;
 166 static receive_list_entry_t *receive_list_tail = NULL;
 167 static pthread_mutex_t       receive_list_lock = PTHREAD_MUTEX_INITIALIZER;
 168 static pthread_cond_t        receive_list_cond = PTHREAD_COND_INITIALIZER;
 169
 170 static struct pollfd *listen_sockets = NULL;
 171 static int listen_sockets_num = 0;
 172
 173 static int listen_loop = 0;
 174 static pthread_t receive_thread_id = 0;
 175 static pthread_t dispatch_thread_id = 0;
 176
 177 static char         send_buffer[BUFF_SIZE];
 178 static char        *send_buffer_ptr;
 179 static int          send_buffer_fill;
 180 static value_list_t send_buffer_vl = VALUE_LIST_STATIC;
 181 static char         send_buffer_type[DATA_MAX_NAME_LEN];
 182 static pthread_mutex_t send_buffer_lock = PTHREAD_MUTEX_INITIALIZER;
 183
 184 static c_avl_tree_t      *cache_tree = NULL;
 185 static pthread_mutex_t  cache_lock = PTHREAD_MUTEX_INITIALIZER;
 186 static time_t           cache_flush_last;
 187 static int              cache_flush_interval = 1800;
 188
 189 /*
 190  * Private functions
 191  */
 192 static int cache_flush (void)
 193 {
 194         char **keys = NULL;
 195         int    keys_num = 0;
 196
 197         char **tmp;
 198         int    i;
 199
 200         char   *key;
 201         time_t *value;
 202         c_avl_iterator_t *iter;
 203
 204         time_t curtime = time (NULL);
 205
 206         iter = c_avl_get_iterator (cache_tree);
 207         while (c_avl_iterator_next (iter, (void *) &key, (void *) &value) == 0)
 208         {
 209                 if ((curtime - *value) <= cache_flush_interval)
 210                         continue;
 211                 tmp = (char **) realloc (keys,
 212                                 (keys_num + 1) * sizeof (char *));
 213                 if (tmp == NULL)
 214                 {
 215                         sfree (keys);
 216                         c_avl_iterator_destroy (iter);
 217                         ERROR ("network plugin: cache_flush: realloc"
 218                                         " failed.");
 219                         return (-1);
 220                 }
 221                 keys = tmp;
 222                 keys[keys_num] = key;
 223                 keys_num++;
 224         } /* while (c_avl_iterator_next) */
 225         c_avl_iterator_destroy (iter);
 226
 227         for (i = 0; i < keys_num; i++)
 228         {
 229                 if (c_avl_remove (cache_tree, keys[i], (void *) &key,
 230                                         (void *) &value) != 0)
 231                 {
 232                         WARNING ("network plugin: cache_flush: c_avl_remove"
 233                                         " (%s) failed.", keys[i]);
 234                         continue;
 235                 }
 236
 237                 sfree (key);
 238                 sfree (value);
 239         }
 240
 241         sfree (keys);
 242
 243         DEBUG ("network plugin: cache_flush: Removed %i %s",
 244                         keys_num, (keys_num == 1) ? "entry" : "entries");
 245         cache_flush_last = curtime;
 246         return (0);
 247 } /* int cache_flush */
 248
 249 static int cache_check (const char *type, const value_list_t *vl)
 250 {
 251         char key[1024];
 252         time_t *value = NULL;
 253         int retval = -1;
 254
 255         if (cache_tree == NULL)
 256                 return (-1);
 257
 258         if (format_name (key, sizeof (key), vl->host, vl->plugin,
 259                                 vl->plugin_instance, type, vl->type_instance))
 260                 return (-1);
 261
 262         pthread_mutex_lock (&cache_lock);
 263
 264         if (c_avl_get (cache_tree, key, (void *) &value) == 0)
 265         {
 266                 if (*value < vl->time)
 267                 {
 268                         *value = vl->time;
 269                         retval = 0;
 270                 }
 271                 else
 272                 {
 273                         DEBUG ("network plugin: cache_check: *value = %i >= vl->time = %i",
 274                                         (int) *value, (int) vl->time);
 275                         retval = 1;
 276                 }
 277         }
 278         else
 279         {
 280                 char *key_copy = strdup (key);
 281                 value = malloc (sizeof (time_t));
 282                 if ((key_copy != NULL) && (value != NULL))
 283                 {
 284                         *value = vl->time;
 285                         c_avl_insert (cache_tree, key_copy, value);
 286                         retval = 0;
 287                 }
 288                 else
 289                 {
 290                         sfree (key_copy);
 291                         sfree (value);
 292                 }
 293         }
 294
 295         if ((time (NULL) - cache_flush_last) > cache_flush_interval)
 296                 cache_flush ();
 297
 298         pthread_mutex_unlock (&cache_lock);
 299
 300         DEBUG ("network plugin: cache_check: key = %s; time = %i; retval = %i",
 301                         key, (int) vl->time, retval);
 302
 303         return (retval);
 304 } /* int cache_check */
 305
 306 static int write_part_values (char **ret_buffer, int *ret_buffer_len,
 307                 const data_set_t *ds, const value_list_t *vl)
 308 {
 309         char *packet_ptr;
 310         int packet_len;
 311         int num_values;
 312
 313         part_header_t pkg_ph;
 314         uint16_t      pkg_num_values;
 315         uint8_t      *pkg_values_types;
 316         value_t      *pkg_values;
 317
 318         int offset;
 319         int i;
 320
 321         num_values = vl->values_len;
 322         packet_len = sizeof (part_header_t) + sizeof (uint16_t)
 323                 + (num_values * sizeof (uint8_t))
 324                 + (num_values * sizeof (value_t));
 325
 326         if (*ret_buffer_len < packet_len)
 327                 return (-1);
 328
 329         pkg_values_types = (uint8_t *) malloc (num_values * sizeof (uint8_t));
 330         if (pkg_values_types == NULL)
 331         {
 332                 ERROR ("network plugin: write_part_values: malloc failed.");
 333                 return (-1);
 334         }
 335
 336         pkg_values = (value_t *) malloc (num_values * sizeof (value_t));
 337         if (pkg_values == NULL)
 338         {
 339                 free (pkg_values_types);
 340                 ERROR ("network plugin: write_part_values: malloc failed.");
 341                 return (-1);
 342         }
 343
 344         pkg_ph.type = htons (TYPE_VALUES);
 345         pkg_ph.length = htons (packet_len);
 346
 347         pkg_num_values = htons ((uint16_t) vl->values_len);
 348
 349         for (i = 0; i < num_values; i++)
 350         {
 351                 if (ds->ds[i].type == DS_TYPE_COUNTER)
 352                 {
 353                         pkg_values_types[i] = DS_TYPE_COUNTER;
 354                         pkg_values[i].counter = htonll (vl->values[i].counter);
 355                 }
 356                 else
 357                 {
 358                         pkg_values_types[i] = DS_TYPE_GAUGE;
 359                         pkg_values[i].gauge = htond (vl->values[i].gauge);
 360                 }
 361         }
 362
 363         /*
 364          * Use `memcpy' to write everything to the buffer, because the pointer
 365          * may be unaligned and some architectures, such as SPARC, can't handle
 366          * that.
 367          */
 368         packet_ptr = *ret_buffer;
 369         offset = 0;
 370         memcpy (packet_ptr + offset, &pkg_ph, sizeof (pkg_ph));
 371         offset += sizeof (pkg_ph);
 372         memcpy (packet_ptr + offset, &pkg_num_values, sizeof (pkg_num_values));
 373         offset += sizeof (pkg_num_values);
 374         memcpy (packet_ptr + offset, pkg_values_types, num_values * sizeof (uint8_t));
 375         offset += num_values * sizeof (uint8_t);
 376         memcpy (packet_ptr + offset, pkg_values, num_values * sizeof (value_t));
 377         offset += num_values * sizeof (value_t);
 378
 379         assert (offset == packet_len);
 380
 381         *ret_buffer = packet_ptr + packet_len;
 382         *ret_buffer_len -= packet_len;
 383
 384         free (pkg_values_types);
 385         free (pkg_values);
 386
 387         return (0);
 388 } /* int write_part_values */
 389
 390 static int write_part_number (char **ret_buffer, int *ret_buffer_len,
 391                 int type, uint64_t value)
 392 {
 393         char *packet_ptr;
 394         int packet_len;
 395
 396         part_header_t pkg_head;
 397         uint64_t pkg_value;
 398
 399         int offset;
 400
 401         packet_len = sizeof (pkg_head) + sizeof (pkg_value);
 402
 403         if (*ret_buffer_len < packet_len)
 404                 return (-1);
 405
 406         pkg_head.type = htons (type);
 407         pkg_head.length = htons (packet_len);
 408         pkg_value = htonll (value);
 409
 410         packet_ptr = *ret_buffer;
 411         offset = 0;
 412         memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head));
 413         offset += sizeof (pkg_head);
 414         memcpy (packet_ptr + offset, &pkg_value, sizeof (pkg_value));
 415         offset += sizeof (pkg_value);
 416
 417         assert (offset == packet_len);
 418
 419         *ret_buffer = packet_ptr + packet_len;
 420         *ret_buffer_len -= packet_len;
 421
 422         return (0);
 423 } /* int write_part_number */
 424
 425 static int write_part_string (char **ret_buffer, int *ret_buffer_len,
 426                 int type, const char *str, int str_len)
 427 {
 428         char *packet_ptr;
 429         int packet_len;
 430
 431         part_header_t pkg_head;
 432
 433         int offset;
 434
 435         packet_len = sizeof (pkg_head) + str_len + 1;
 436         if (*ret_buffer_len < packet_len)
 437                 return (-1);
 438
 439         pkg_head.type = htons (type);
 440         pkg_head.length = htons (packet_len);
 441
 442         packet_ptr = *ret_buffer;
 443         offset = 0;
 444         memcpy (packet_ptr + offset, &pkg_head, sizeof (pkg_head));
 445         offset += sizeof (pkg_head);
 446         memcpy (packet_ptr + offset, str, str_len);
 447         offset += str_len;
 448         memset (packet_ptr + offset, '\0', 1);
 449         offset += 1;
 450
 451         assert (offset == packet_len);
 452
 453         *ret_buffer = packet_ptr + packet_len;
 454         *ret_buffer_len -= packet_len;
 455
 456         return (0);
 457 } /* int write_part_string */
 458
 459 static int parse_part_values (void **ret_buffer, int *ret_buffer_len,
 460                 value_t **ret_values, int *ret_num_values)
 461 {
 462         char *buffer = *ret_buffer;
 463         int   buffer_len = *ret_buffer_len;
 464         part_values_t pv;
 465         int   i;
 466
 467         uint16_t h_length;
 468         uint16_t h_type;
 469         uint16_t h_num;
 470
 471         if (buffer_len < (15))
 472         {
 473                 DEBUG ("network plugin: packet is too short: buffer_len = %i",
 474                                 buffer_len);
 475                 return (-1);
 476         }
 477
 478         pv.head = (part_header_t *) buffer;
 479         h_length = ntohs (pv.head->length);
 480         h_type = ntohs (pv.head->type);
 481
 482         assert (h_type == TYPE_VALUES);
 483
 484         pv.num_values = (uint16_t *) (pv.head + 1);
 485         h_num = ntohs (*pv.num_values);
 486
 487         if (h_num != ((h_length - 6) / 9))
 488         {
 489                 DEBUG ("`length' and `num of values' don't match");
 490                 return (-1);
 491         }
 492
 493         pv.values_types = (uint8_t *) (pv.num_values + 1);
 494         pv.values = (value_t *) (pv.values_types + h_num);
 495
 496         for (i = 0; i < h_num; i++)
 497                 if (pv.values_types[i] == DS_TYPE_COUNTER)
 498                         pv.values[i].counter = ntohll (pv.values[i].counter);
 499                 else
 500                         pv.values[i].gauge = ntohd (pv.values[i].gauge);
 501
 502         *ret_buffer     = (void *) (pv.values + h_num);
 503         *ret_buffer_len = buffer_len - h_length;
 504         *ret_num_values = h_num;
 505         *ret_values     = pv.values;
 506
 507         return (0);
 508 } /* int parse_part_values */
 509
 510 static int parse_part_number (void **ret_buffer, int *ret_buffer_len,
 511                 uint64_t *value)
 512 {
 513         part_number_t pn;
 514         uint16_t len;
 515
 516         pn.head = (part_header_t *) *ret_buffer;
 517         pn.value = (uint64_t *) (pn.head + 1);
 518
 519         len = ntohs (pn.head->length);
 520         if (len != 12)
 521                 return (-1);
 522         if (len > *ret_buffer_len)
 523                 return (-1);
 524         *value = ntohll (*pn.value);
 525
 526         *ret_buffer = (void *) (pn.value + 1);
 527         *ret_buffer_len -= len;
 528
 529         return (0);
 530 } /* int parse_part_number */
 531
 532 static int parse_part_string (void **ret_buffer, int *ret_buffer_len,
 533                 char *output, int output_len)
 534 {
 535         char *buffer = *ret_buffer;
 536         int   buffer_len = *ret_buffer_len;
 537         part_string_t ps;
 538
 539         uint16_t h_length;
 540         uint16_t h_type;
 541
 542         DEBUG ("network plugin: parse_part_string: ret_buffer = %p;"
 543                         " ret_buffer_len = %i; output = %p; output_len = %i;",
 544                         *ret_buffer, *ret_buffer_len,
 545                         (void *) output, output_len);
 546
 547         ps.head = (part_header_t *) buffer;
 548
 549         h_length = ntohs (ps.head->length);
 550         h_type = ntohs (ps.head->type);
 551
 552         DEBUG ("network plugin: parse_part_string: length = %hu; type = %hu;",
 553                         h_length, h_type);
 554
 555         if (buffer_len < h_length)
 556         {
 557                 DEBUG ("packet is too short");
 558                 return (-1);
 559         }
 560         assert ((h_type == TYPE_HOST)
 561                         || (h_type == TYPE_PLUGIN)
 562                         || (h_type == TYPE_PLUGIN_INSTANCE)
 563                         || (h_type == TYPE_TYPE)
 564                         || (h_type == TYPE_TYPE_INSTANCE));
 565
 566         ps.value = buffer + 4;
 567         if (ps.value[h_length - 5] != '\0')
 568         {
 569                 DEBUG ("String does not end with a nullbyte");
 570                 return (-1);
 571         }
 572
 573         if (output_len < (h_length - 4))
 574         {
 575                 DEBUG ("output buffer is too small");
 576                 return (-1);
 577         }
 578         strcpy (output, ps.value);
 579
 580         DEBUG ("network plugin: parse_part_string: output = %s", output);
 581
 582         *ret_buffer = (void *) (buffer + h_length);
 583         *ret_buffer_len = buffer_len - h_length;
 584
 585         return (0);
 586 } /* int parse_part_string */
 587
 588 static int parse_packet (void *buffer, int buffer_len)
 589 {
 590         part_header_t *header;
 591         int status;
 592
 593         value_list_t vl = VALUE_LIST_INIT;
 594         char type[DATA_MAX_NAME_LEN];
 595
 596         DEBUG ("network plugin: parse_packet: buffer = %p; buffer_len = %i;",
 597                         buffer, buffer_len);
 598
 599         memset (&vl, '\0', sizeof (vl));
 600         memset (&type, '\0', sizeof (type));
 601         status = 0;
 602
 603         while ((status == 0) && (buffer_len > sizeof (part_header_t)))
 604         {
 605                 header = (part_header_t *) buffer;
 606
 607                 if (ntohs (header->length) > buffer_len)
 608                         break;
 609                 /* Assure that this loop terminates eventually */
 610                 if (ntohs (header->length) < 4)
 611                         break;
 612
 613                 if (ntohs (header->type) == TYPE_VALUES)
 614                 {
 615                         status = parse_part_values (&buffer, &buffer_len,
 616                                         &vl.values, &vl.values_len);
 617
 618                         if (status != 0)
 619                         {
 620                                 DEBUG ("parse_part_values failed.");
 621                                 break;
 622                         }
 623
 624                         if ((vl.time > 0)
 625                                         && (strlen (vl.host) > 0)
 626                                         && (strlen (vl.plugin) > 0)
 627                                         && (strlen (type) > 0)
 628                                         && (cache_check (type, &vl) == 0))
 629                         {
 630                                 DEBUG ("network plugin: parse_packet:"
 631                                                 " dispatching values");
 632                                 plugin_dispatch_values (type, &vl);
 633                         }
 634                         else
 635                         {
 636                                 DEBUG ("network plugin: parse_packet:"
 637                                                 " NOT dispatching values");
 638                         }
 639                 }
 640                 else if (ntohs (header->type) == TYPE_TIME)
 641                 {
 642                         uint64_t tmp = 0;
 643                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 644                         if (status == 0)
 645                                 vl.time = (time_t) tmp;
 646                 }
 647                 else if (ntohs (header->type) == TYPE_INTERVAL)
 648                 {
 649                         uint64_t tmp = 0;
 650                         status = parse_part_number (&buffer, &buffer_len, &tmp);
 651                         if (status == 0)
 652                                 vl.interval = (int) tmp;
 653                 }
 654                 else if (ntohs (header->type) == TYPE_HOST)
 655                 {
 656                         status = parse_part_string (&buffer, &buffer_len,
 657                                         vl.host, sizeof (vl.host));
 658                         DEBUG ("network plugin: parse_packet: vl.host = %s", vl.host);
 659                 }
 660                 else if (ntohs (header->type) == TYPE_PLUGIN)
 661                 {
 662                         status = parse_part_string (&buffer, &buffer_len,
 663                                         vl.plugin, sizeof (vl.plugin));
 664                         DEBUG ("network plugin: parse_packet: vl.plugin = %s", vl.plugin);
 665                 }
 666                 else if (ntohs (header->type) == TYPE_PLUGIN_INSTANCE)
 667                 {
 668                         status = parse_part_string (&buffer, &buffer_len,
 669                                         vl.plugin_instance, sizeof (vl.plugin_instance));
 670                         DEBUG ("network plugin: parse_packet: vl.plugin_instance = %s", vl.plugin_instance);
 671                 }
 672                 else if (ntohs (header->type) == TYPE_TYPE)
 673                 {
 674                         status = parse_part_string (&buffer, &buffer_len,
 675                                         type, sizeof (type));
 676                         DEBUG ("network plugin: parse_packet: type = %s", type);
 677                 }
 678                 else if (ntohs (header->type) == TYPE_TYPE_INSTANCE)
 679                 {
 680                         status = parse_part_string (&buffer, &buffer_len,
 681                                         vl.type_instance, sizeof (vl.type_instance));
 682                         DEBUG ("network plugin: parse_packet: vl.type_instance = %s", vl.type_instance);
 683                 }
 684                 else
 685                 {
 686                         DEBUG ("network plugin: parse_packet: Unknown part"
 687                                         " type: 0x%0hx", ntohs (header->type));
 688                         buffer = ((char *) buffer) + ntohs (header->length);
 689                 }
 690         } /* while (buffer_len > sizeof (part_header_t)) */
 691
 692         return (0);
 693 } /* int parse_packet */
 694
 695 static void free_sockent (sockent_t *se)
 696 {
 697         sockent_t *next;
 698         while (se != NULL)
 699         {
 700                 next = se->next;
 701                 free (se->addr);
 702                 free (se);
 703                 se = next;
 704         }
 705 } /* void free_sockent */
 706
 707 /*
 708  * int network_set_ttl
 709  *
 710  * Set the `IP_MULTICAST_TTL', `IP_TTL', `IPV6_MULTICAST_HOPS' or
 711  * `IPV6_UNICAST_HOPS', depending on which option is applicable.
 712  *
 713  * The `struct addrinfo' is used to destinguish between unicast and multicast
 714  * sockets.
 715  */
 716 static int network_set_ttl (const sockent_t *se, const struct addrinfo *ai)
 717 {
 718         if ((network_config_ttl < 1) || (network_config_ttl > 255))
 719                 return (-1);
 720
 721         DEBUG ("ttl = %i", network_config_ttl);
 722
 723         if (ai->ai_family == AF_INET)
 724         {
 725                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 726                 int optname;
 727
 728                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 729                         optname = IP_MULTICAST_TTL;
 730                 else
 731                         optname = IP_TTL;
 732
 733                 if (setsockopt (se->fd, IPPROTO_IP, optname,
 734                                         &network_config_ttl,
 735                                         sizeof (network_config_ttl)) == -1)
 736                 {
 737                         char errbuf[1024];
 738                         ERROR ("setsockopt: %s",
 739                                         sstrerror (errno, errbuf, sizeof (errbuf)));
 740                         return (-1);
 741                 }
 742         }
 743         else if (ai->ai_family == AF_INET6)
 744         {
 745                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 746                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 747                 int optname;
 748
 749                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 750                         optname = IPV6_MULTICAST_HOPS;
 751                 else
 752                         optname = IPV6_UNICAST_HOPS;
 753
 754                 if (setsockopt (se->fd, IPPROTO_IPV6, optname,
 755                                         &network_config_ttl,
 756                                         sizeof (network_config_ttl)) == -1)
 757                 {
 758                         char errbuf[1024];
 759                         ERROR ("setsockopt: %s",
 760                                         sstrerror (errno, errbuf,
 761                                                 sizeof (errbuf)));
 762                         return (-1);
 763                 }
 764         }
 765
 766         return (0);
 767 } /* int network_set_ttl */
 768
 769 static int network_bind_socket (const sockent_t *se, const struct addrinfo *ai)
 770 {
 771         int loop = 0;
 772
 773         DEBUG ("fd = %i; calling `bind'", se->fd);
 774
 775         if (bind (se->fd, ai->ai_addr, ai->ai_addrlen) == -1)
 776         {
 777                 char errbuf[1024];
 778                 ERROR ("bind: %s",
 779                                 sstrerror (errno, errbuf, sizeof (errbuf)));
 780                 return (-1);
 781         }
 782
 783         if (ai->ai_family == AF_INET)
 784         {
 785                 struct sockaddr_in *addr = (struct sockaddr_in *) ai->ai_addr;
 786                 if (IN_MULTICAST (ntohl (addr->sin_addr.s_addr)))
 787                 {
 788                         struct ip_mreq mreq;
 789
 790                         DEBUG ("fd = %i; IPv4 multicast address found", se->fd);
 791
 792                         mreq.imr_multiaddr.s_addr = addr->sin_addr.s_addr;
 793                         mreq.imr_interface.s_addr = htonl (INADDR_ANY);
 794
 795                         if (setsockopt (se->fd, IPPROTO_IP, IP_MULTICAST_LOOP,
 796                                                 &loop, sizeof (loop)) == -1)
 797                         {
 798                                 char errbuf[1024];
 799                                 ERROR ("setsockopt: %s",
 800                                                 sstrerror (errno, errbuf,
 801                                                         sizeof (errbuf)));
 802                                 return (-1);
 803                         }
 804
 805                         if (setsockopt (se->fd, IPPROTO_IP, IP_ADD_MEMBERSHIP,
 806                                                 &mreq, sizeof (mreq)) == -1)
 807                         {
 808                                 char errbuf[1024];
 809                                 ERROR ("setsockopt: %s",
 810                                                 sstrerror (errno, errbuf,
 811                                                         sizeof (errbuf)));
 812                                 return (-1);
 813                         }
 814                 }
 815         }
 816         else if (ai->ai_family == AF_INET6)
 817         {
 818                 /* Useful example: http://gsyc.escet.urjc.es/~eva/IPv6-web/examples/mcast.html */
 819                 struct sockaddr_in6 *addr = (struct sockaddr_in6 *) ai->ai_addr;
 820                 if (IN6_IS_ADDR_MULTICAST (&addr->sin6_addr))
 821                 {
 822                         struct ipv6_mreq mreq;
 823
 824                         DEBUG ("fd = %i; IPv6 multicast address found", se->fd);
 825
 826                         memcpy (&mreq.ipv6mr_multiaddr,
 827                                         &addr->sin6_addr,
 828                                         sizeof (addr->sin6_addr));
 829
 830                         /* http://developer.apple.com/documentation/Darwin/Reference/ManPages/man4/ip6.4.html
 831                          * ipv6mr_interface may be set to zeroes to
 832                          * choose the default multicast interface or to
 833                          * the index of a particular multicast-capable
 834                          * interface if the host is multihomed.
 835                          * Membership is associ-associated with a
 836                          * single interface; programs running on
 837                          * multihomed hosts may need to join the same
 838                          * group on more than one interface.*/
 839                         mreq.ipv6mr_interface = 0;
 840
 841                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_MULTICAST_LOOP,
 842                                                 &loop, sizeof (loop)) == -1)
 843                         {
 844                                 char errbuf[1024];
 845                                 ERROR ("setsockopt: %s",
 846                                                 sstrerror (errno, errbuf,
 847                                                         sizeof (errbuf)));
 848                                 return (-1);
 849                         }
 850
 851                         if (setsockopt (se->fd, IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP,
 852                                                 &mreq, sizeof (mreq)) == -1)
 853                         {
 854                                 char errbuf[1024];
 855                                 ERROR ("setsockopt: %s",
 856                                                 sstrerror (errno, errbuf,
 857                                                         sizeof (errbuf)));
 858                                 return (-1);
 859                         }
 860                 }
 861         }
 862
 863         return (0);
 864 } /* int network_bind_socket */
 865
 866 static sockent_t *network_create_socket (const char *node,
 867                 const char *service,
 868                 int listen)
 869 {
 870         struct addrinfo  ai_hints;
 871         struct addrinfo *ai_list, *ai_ptr;
 872         int              ai_return;
 873
 874         sockent_t *se_head = NULL;
 875         sockent_t *se_tail = NULL;
 876
 877         DEBUG ("node = %s, service = %s", node, service);
 878
 879         memset (&ai_hints, '\0', sizeof (ai_hints));
 880         ai_hints.ai_flags    = 0;
 881 #ifdef AI_PASSIVE
 882         ai_hints.ai_flags |= AI_PASSIVE;
 883 #endif
 884 #ifdef AI_ADDRCONFIG
 885         ai_hints.ai_flags |= AI_ADDRCONFIG;
 886 #endif
 887         ai_hints.ai_family   = AF_UNSPEC;
 888         ai_hints.ai_socktype = SOCK_DGRAM;
 889         ai_hints.ai_protocol = IPPROTO_UDP;
 890
 891         ai_return = getaddrinfo (node, service, &ai_hints, &ai_list);
 892         if (ai_return != 0)
 893         {
 894                 char errbuf[1024];
 895                 ERROR ("getaddrinfo (%s, %s): %s",
 896                                 (node == NULL) ? "(null)" : node,
 897                                 (service == NULL) ? "(null)" : service,
 898                                 (ai_return == EAI_SYSTEM)
 899                                 ? sstrerror (errno, errbuf, sizeof (errbuf))
 900                                 : gai_strerror (ai_return));
 901                 return (NULL);
 902         }
 903
 904         for (ai_ptr = ai_list; ai_ptr != NULL; ai_ptr = ai_ptr->ai_next)
 905         {
 906                 sockent_t *se;
 907
 908                 if ((se = (sockent_t *) malloc (sizeof (sockent_t))) == NULL)
 909                 {
 910                         char errbuf[1024];
 911                         ERROR ("malloc: %s",
 912                                         sstrerror (errno, errbuf,
 913                                                 sizeof (errbuf)));
 914                         continue;
 915                 }
 916
 917                 if ((se->addr = (struct sockaddr_storage *) malloc (sizeof (struct sockaddr_storage))) == NULL)
 918                 {
 919                         char errbuf[1024];
 920                         ERROR ("malloc: %s",
 921                                         sstrerror (errno, errbuf,
 922                                                 sizeof (errbuf)));
 923                         free (se);
 924                         continue;
 925                 }
 926
 927                 assert (sizeof (struct sockaddr_storage) >= ai_ptr->ai_addrlen);
 928                 memset (se->addr, '\0', sizeof (struct sockaddr_storage));
 929                 memcpy (se->addr, ai_ptr->ai_addr, ai_ptr->ai_addrlen);
 930                 se->addrlen = ai_ptr->ai_addrlen;
 931
 932                 se->fd   = socket (ai_ptr->ai_family,
 933                                 ai_ptr->ai_socktype,
 934                                 ai_ptr->ai_protocol);
 935                 se->next = NULL;
 936
 937                 if (se->fd == -1)
 938                 {
 939                         char errbuf[1024];
 940                         ERROR ("socket: %s",
 941                                         sstrerror (errno, errbuf,
 942                                                 sizeof (errbuf)));
 943                         free (se->addr);
 944                         free (se);
 945                         continue;
 946                 }
 947
 948                 if (listen != 0)
 949                 {
 950                         if (network_bind_socket (se, ai_ptr) != 0)
 951                         {
 952                                 close (se->fd);
 953                                 free (se->addr);
 954                                 free (se);
 955                                 continue;
 956                         }
 957                 }
 958                 else /* listen == 0 */
 959                 {
 960                         network_set_ttl (se, ai_ptr);
 961                 }
 962
 963                 if (se_tail == NULL)
 964                 {
 965                         se_head = se;
 966                         se_tail = se;
 967                 }
 968                 else
 969                 {
 970                         se_tail->next = se;
 971                         se_tail = se;
 972                 }
 973
 974                 /* We don't open more than one write-socket per node/service pair.. */
 975                 if (listen == 0)
 976                         break;
 977         }
 978
 979         freeaddrinfo (ai_list);
 980
 981         return (se_head);
 982 } /* sockent_t *network_create_socket */
 983
 984 static sockent_t *network_create_default_socket (int listen)
 985 {
 986         sockent_t *se_ptr  = NULL;
 987         sockent_t *se_head = NULL;
 988         sockent_t *se_tail = NULL;
 989
 990         se_ptr = network_create_socket (NET_DEFAULT_V6_ADDR,
 991                         NET_DEFAULT_PORT, listen);
 992
 993         /* Don't send to the same machine in IPv6 and IPv4 if both are available. */
 994         if ((listen == 0) && (se_ptr != NULL))
 995                 return (se_ptr);
 996
 997         if (se_ptr != NULL)
 998         {
 999                 se_head = se_ptr;
1000                 se_tail = se_ptr;
1001                 while (se_tail->next != NULL)
1002                         se_tail = se_tail->next;
1003         }
1004
1005         se_ptr = network_create_socket (NET_DEFAULT_V4_ADDR, NET_DEFAULT_PORT, listen);
1006
1007         if (se_tail == NULL)
1008                 return (se_ptr);
1009
1010         se_tail->next = se_ptr;
1011         return (se_head);
1012 } /* sockent_t *network_create_default_socket */
1013
1014 static int network_add_listen_socket (const char *node, const char *service)
1015 {
1016         sockent_t *se;
1017         sockent_t *se_ptr;
1018         int se_num = 0;
1019
1020         if (service == NULL)
1021                 service = NET_DEFAULT_PORT;
1022
1023         if (node == NULL)
1024                 se = network_create_default_socket (1 /* listen == true */);
1025         else
1026                 se = network_create_socket (node, service, 1 /* listen == true */);
1027
1028         if (se == NULL)
1029                 return (-1);
1030
1031         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
1032                 se_num++;
1033
1034         listen_sockets = (struct pollfd *) realloc (listen_sockets,
1035                         (listen_sockets_num + se_num)
1036                         * sizeof (struct pollfd));
1037
1038         for (se_ptr = se; se_ptr != NULL; se_ptr = se_ptr->next)
1039         {
1040                 listen_sockets[listen_sockets_num].fd = se_ptr->fd;
1041                 listen_sockets[listen_sockets_num].events = POLLIN | POLLPRI;
1042                 listen_sockets[listen_sockets_num].revents = 0;
1043                 listen_sockets_num++;
1044         } /* for (se) */
1045
1046         free_sockent (se);
1047         return (0);
1048 } /* int network_add_listen_socket */
1049
1050 static int network_add_sending_socket (const char *node, const char *service)
1051 {
1052         sockent_t *se;
1053         sockent_t *se_ptr;
1054
1055         if (service == NULL)
1056                 service = NET_DEFAULT_PORT;
1057
1058         if (node == NULL)
1059                 se = network_create_default_socket (0 /* listen == false */);
1060         else
1061                 se = network_create_socket (node, service, 0 /* listen == false */);
1062
1063         if (se == NULL)
1064                 return (-1);
1065
1066         if (sending_sockets == NULL)
1067         {
1068                 sending_sockets = se;
1069                 return (0);
1070         }
1071
1072         for (se_ptr = sending_sockets; se_ptr->next != NULL; se_ptr = se_ptr->next)
1073                 /* seek end */;
1074
1075         se_ptr->next = se;
1076         return (0);
1077 } /* int network_get_listen_socket */
1078
1079 static void *dispatch_thread (void *arg)
1080 {
1081   while (42)
1082   {
1083     receive_list_entry_t *ent;
1084
1085     /* Lock and wait for more data to come in */
1086     pthread_mutex_lock (&receive_list_lock);
1087     while ((listen_loop == 0)
1088         && (receive_list_head == NULL))
1089       pthread_cond_wait (&receive_list_cond, &receive_list_lock);
1090
1091     /* Remove the head entry and unlock */
1092     ent = receive_list_head;
1093     if (ent != NULL)
1094       receive_list_head = ent->next;
1095     pthread_mutex_unlock (&receive_list_lock);
1096
1097     /* Check whether we are supposed to exit. We do NOT check `listen_loop'
1098      * because we dispatch all missing packets before shutting down. */
1099     if (ent == NULL)
1100       break;
1101
1102     parse_packet (ent->data, ent->data_len);
1103
1104     sfree (ent);
1105   } /* while (42) */
1106
1107   return (NULL);
1108 } /* void *receive_thread */
1109
1110 static int network_receive (void)
1111 {
1112         char buffer[BUFF_SIZE];
1113         int  buffer_len;
1114
1115         int i;
1116         int status;
1117
1118         if (listen_sockets_num == 0)
1119                 network_add_listen_socket (NULL, NULL);
1120
1121         if (listen_sockets_num == 0)
1122         {
1123                 ERROR ("network: Failed to open a listening socket.");
1124                 return (-1);
1125         }
1126
1127         while (listen_loop == 0)
1128         {
1129                 status = poll (listen_sockets, listen_sockets_num, -1);
1130
1131                 if (status <= 0)
1132                 {
1133                         char errbuf[1024];
1134                         if (errno == EINTR)
1135                                 continue;
1136                         ERROR ("poll failed: %s",
1137                                         sstrerror (errno, errbuf, sizeof (errbuf)));
1138                         return (-1);
1139                 }
1140
1141                 for (i = 0; (i < listen_sockets_num) && (status > 0); i++)
1142                 {
1143                         receive_list_entry_t *ent;
1144
1145                         if ((listen_sockets[i].revents & (POLLIN | POLLPRI)) == 0)
1146                                 continue;
1147                         status--;
1148
1149                         buffer_len = recv (listen_sockets[i].fd,
1150                                         buffer, sizeof (buffer),
1151                                         0 /* no flags */);
1152                         if (buffer_len < 0)
1153                         {
1154                                 char errbuf[1024];
1155                                 ERROR ("recv failed: %s",
1156                                                 sstrerror (errno, errbuf,
1157                                                         sizeof (errbuf)));
1158                                 return (-1);
1159                         }
1160
1161                         ent = malloc (sizeof (receive_list_entry_t));
1162                         if (ent == NULL)
1163                         {
1164                                 ERROR ("network plugin: malloc failed.");
1165                                 return (-1);
1166                         }
1167                         memset (ent, '\0', sizeof (receive_list_entry_t));
1168
1169                         /* Hopefully this be optimized out by the compiler. It
1170                          * might help prevent stupid bugs in the future though.
1171                          */
1172                         assert (sizeof (ent->data) == sizeof (buffer));
1173
1174                         memcpy (ent->data, buffer, buffer_len);
1175                         ent->data_len = buffer_len;
1176
1177                         pthread_mutex_lock (&receive_list_lock);
1178                         if (receive_list_head == NULL)
1179                         {
1180                                 receive_list_head = ent;
1181                                 receive_list_tail = ent;
1182                         }
1183                         else
1184                         {
1185                                 receive_list_tail->next = ent;
1186                                 receive_list_tail = ent;
1187                         }
1188                         pthread_cond_signal (&receive_list_cond);
1189                         pthread_mutex_unlock (&receive_list_lock);
1190                 } /* for (listen_sockets) */
1191         } /* while (listen_loop == 0) */
1192
1193         return (0);
1194 }
1195
1196 static void *receive_thread (void *arg)
1197 {
1198         return (network_receive () ? (void *) 1 : (void *) 0);
1199 } /* void *receive_thread */
1200
1201 static void network_send_buffer (const char *buffer, int buffer_len)
1202 {
1203         sockent_t *se;
1204         int status;
1205
1206         DEBUG ("network plugin: network_send_buffer: buffer_len = %i", buffer_len);
1207
1208         for (se = sending_sockets; se != NULL; se = se->next)
1209         {
1210                 while (42)
1211                 {
1212                         status = sendto (se->fd, buffer, buffer_len, 0 /* no flags */,
1213                                         (struct sockaddr *) se->addr, se->addrlen);
1214                         if (status < 0)
1215                         {
1216                                 char errbuf[1024];
1217                                 if (errno == EINTR)
1218                                         continue;
1219                                 ERROR ("network plugin: sendto failed: %s",
1220                                                 sstrerror (errno, errbuf,
1221                                                         sizeof (errbuf)));
1222                                 break;
1223                         }
1224
1225                         break;
1226                 } /* while (42) */
1227         } /* for (sending_sockets) */
1228 } /* void network_send_buffer */
1229
1230 static int add_to_buffer (char *buffer, int buffer_size,
1231                 value_list_t *vl_def, char *type_def,
1232                 const data_set_t *ds, const value_list_t *vl)
1233 {
1234         char *buffer_orig = buffer;
1235
1236         if (strcmp (vl_def->host, vl->host) != 0)
1237         {
1238                 if (write_part_string (&buffer, &buffer_size, TYPE_HOST,
1239                                         vl->host, strlen (vl->host)) != 0)
1240                         return (-1);
1241                 strcpy (vl_def->host, vl->host);
1242         }
1243
1244         if (vl_def->time != vl->time)
1245         {
1246                 if (write_part_number (&buffer, &buffer_size, TYPE_TIME,
1247                                         (uint64_t) vl->time))
1248                         return (-1);
1249                 vl_def->time = vl->time;
1250         }
1251
1252         if (vl_def->interval != vl->interval)
1253         {
1254                 if (write_part_number (&buffer, &buffer_size, TYPE_INTERVAL,
1255                                         (uint64_t) vl->interval))
1256                         return (-1);
1257                 vl_def->interval = vl->interval;
1258         }
1259
1260         if (strcmp (vl_def->plugin, vl->plugin) != 0)
1261         {
1262                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN,
1263                                         vl->plugin, strlen (vl->plugin)) != 0)
1264                         return (-1);
1265                 strcpy (vl_def->plugin, vl->plugin);
1266         }
1267
1268         if (strcmp (vl_def->plugin_instance, vl->plugin_instance) != 0)
1269         {
1270                 if (write_part_string (&buffer, &buffer_size, TYPE_PLUGIN_INSTANCE,
1271                                         vl->plugin_instance,
1272                                         strlen (vl->plugin_instance)) != 0)
1273                         return (-1);
1274                 strcpy (vl_def->plugin_instance, vl->plugin_instance);
1275         }
1276
1277         if (strcmp (type_def, ds->type) != 0)
1278         {
1279                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE,
1280                                         ds->type, strlen (ds->type)) != 0)
1281                         return (-1);
1282                 strcpy (type_def, ds->type);
1283         }
1284
1285         if (strcmp (vl_def->type_instance, vl->type_instance) != 0)
1286         {
1287                 if (write_part_string (&buffer, &buffer_size, TYPE_TYPE_INSTANCE,
1288                                         vl->type_instance,
1289                                         strlen (vl->type_instance)) != 0)
1290                         return (-1);
1291                 strcpy (vl_def->type_instance, vl->type_instance);
1292         }
1293
1294         if (write_part_values (&buffer, &buffer_size, ds, vl) != 0)
1295                 return (-1);
1296
1297         return (buffer - buffer_orig);
1298 } /* int add_to_buffer */
1299
1300 static void flush_buffer (void)
1301 {
1302         DEBUG ("network plugin: flush_buffer: send_buffer_fill = %i",
1303                         send_buffer_fill);
1304
1305         network_send_buffer (send_buffer, send_buffer_fill);
1306         send_buffer_ptr  = send_buffer;
1307         send_buffer_fill = 0;
1308         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1309         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1310 }
1311
1312 static int network_write (const data_set_t *ds, const value_list_t *vl)
1313 {
1314         int status;
1315
1316         /* If the value is already in the cache, we have received it via the
1317          * network. We write it again if forwarding is activated. It's then in
1318          * the cache and should we receive it again we will ignore it. */
1319         status = cache_check (ds->type, vl);
1320         if ((network_config_forward == 0)
1321                         && (status != 0))
1322                 return (0);
1323
1324         pthread_mutex_lock (&send_buffer_lock);
1325
1326         status = add_to_buffer (send_buffer_ptr,
1327                         sizeof (send_buffer) - send_buffer_fill,
1328                         &send_buffer_vl, send_buffer_type,
1329                         ds, vl);
1330         if (status >= 0)
1331         {
1332                 /* status == bytes added to the buffer */
1333                 send_buffer_fill += status;
1334                 send_buffer_ptr  += status;
1335         }
1336         else
1337         {
1338                 flush_buffer ();
1339
1340                 status = add_to_buffer (send_buffer_ptr,
1341                                 sizeof (send_buffer) - send_buffer_fill,
1342                                 &send_buffer_vl, send_buffer_type,
1343                                 ds, vl);
1344
1345                 if (status >= 0)
1346                 {
1347                         send_buffer_fill += status;
1348                         send_buffer_ptr  += status;
1349                 }
1350         }
1351
1352         if (status < 0)
1353         {
1354                 ERROR ("network plugin: Unable to append to the "
1355                                 "buffer for some weird reason");
1356         }
1357         else if ((sizeof (send_buffer) - send_buffer_fill) < 15)
1358         {
1359                 flush_buffer ();
1360         }
1361
1362         pthread_mutex_unlock (&send_buffer_lock);
1363
1364         return ((status < 0) ? -1 : 0);
1365 } /* int network_write */
1366
1367 static int network_config (const char *key, const char *val)
1368 {
1369         char *node;
1370         char *service;
1371
1372         char *fields[3];
1373         int   fields_num;
1374
1375         if ((strcasecmp ("Listen", key) == 0)
1376                         || (strcasecmp ("Server", key) == 0))
1377         {
1378                 char *val_cpy = strdup (val);
1379                 if (val_cpy == NULL)
1380                         return (1);
1381
1382                 service = NET_DEFAULT_PORT;
1383                 fields_num = strsplit (val_cpy, fields, 3);
1384                 if ((fields_num != 1)
1385                                 && (fields_num != 2))
1386                         return (1);
1387                 else if (fields_num == 2)
1388                 {
1389                         if ((service = strchr (fields[1], '.')) != NULL)
1390                                 *service = '\0';
1391                         service = fields[1];
1392                 }
1393                 node = fields[0];
1394
1395                 if (strcasecmp ("Listen", key) == 0)
1396                         network_add_listen_socket (node, service);
1397                 else
1398                         network_add_sending_socket (node, service);
1399         }
1400         else if (strcasecmp ("TimeToLive", key) == 0)
1401         {
1402                 int tmp = atoi (val);
1403                 if ((tmp > 0) && (tmp < 256))
1404                         network_config_ttl = tmp;
1405                 else
1406                         return (1);
1407         }
1408         else if (strcasecmp ("Forward", key) == 0)
1409         {
1410                 if ((strcasecmp ("true", val) == 0)
1411                                 || (strcasecmp ("yes", val) == 0)
1412                                 || (strcasecmp ("on", val) == 0))
1413                         network_config_forward = 1;
1414                 else
1415                         network_config_forward = 0;
1416         }
1417         else if (strcasecmp ("CacheFlush", key) == 0)
1418         {
1419                 int tmp = atoi (val);
1420                 if (tmp > 0)
1421                         cache_flush_interval = tmp;
1422                 else return (1);
1423         }
1424         else
1425         {
1426                 return (-1);
1427         }
1428         return (0);
1429 } /* int network_config */
1430
1431 static int network_shutdown (void)
1432 {
1433         listen_loop++;
1434
1435         /* Kill the listening thread */
1436         if (receive_thread_id != (pthread_t) 0)
1437         {
1438                 pthread_kill (receive_thread_id, SIGTERM);
1439                 pthread_join (receive_thread_id, NULL /* no return value */);
1440                 receive_thread_id = (pthread_t) 0;
1441         }
1442
1443         /* Shutdown the dispatching thread */
1444         if (dispatch_thread_id != (pthread_t) 0)
1445                 pthread_cond_broadcast (&receive_list_cond);
1446
1447         if (send_buffer_fill > 0)
1448                 flush_buffer ();
1449
1450         if (cache_tree != NULL)
1451         {
1452                 void *key;
1453                 void *value;
1454
1455                 while (c_avl_pick (cache_tree, &key, &value) == 0)
1456                 {
1457                         sfree (key);
1458                         sfree (value);
1459                 }
1460                 c_avl_destroy (cache_tree);
1461                 cache_tree = NULL;
1462         }
1463
1464         /* TODO: Close `sending_sockets' */
1465
1466         plugin_unregister_config ("network");
1467         plugin_unregister_init ("network");
1468         plugin_unregister_write ("network");
1469         plugin_unregister_shutdown ("network");
1470
1471         return (0);
1472 } /* int network_shutdown */
1473
1474 static int network_init (void)
1475 {
1476         plugin_register_shutdown ("network", network_shutdown);
1477
1478         send_buffer_ptr  = send_buffer;
1479         send_buffer_fill = 0;
1480         memset (&send_buffer_vl, '\0', sizeof (send_buffer_vl));
1481         memset (send_buffer_type, '\0', sizeof (send_buffer_type));
1482
1483         cache_tree = c_avl_create ((int (*) (const void *, const void *)) strcmp);
1484         cache_flush_last = time (NULL);
1485
1486         /* setup socket(s) and so on */
1487         if (sending_sockets != NULL)
1488                 plugin_register_write ("network", network_write);
1489
1490         if ((listen_sockets_num != 0) && (receive_thread_id == 0))
1491         {
1492                 int status;
1493
1494                 status = pthread_create (&dispatch_thread_id,
1495                                 NULL /* no attributes */,
1496                                 dispatch_thread,
1497                                 NULL /* no argument */);
1498                 if (status != 0)
1499                 {
1500                         char errbuf[1024];
1501                         ERROR ("network: pthread_create failed: %s",
1502                                         sstrerror (errno, errbuf,
1503                                                 sizeof (errbuf)));
1504                 }
1505
1506                 status = pthread_create (&receive_thread_id,
1507                                 NULL /* no attributes */,
1508                                 receive_thread,
1509                                 NULL /* no argument */);
1510                 if (status != 0)
1511                 {
1512                         char errbuf[1024];
1513                         ERROR ("network: pthread_create failed: %s",
1514                                         sstrerror (errno, errbuf,
1515                                                 sizeof (errbuf)));
1516                 }
1517         }
1518         return (0);
1519 } /* int network_init */
1520
1521 void module_register (void)
1522 {
1523         plugin_register_config ("network", network_config,
1524                         config_keys, config_keys_num);
1525         plugin_register_init   ("network", network_init);
1526 } /* void module_register */