src/email.c

   1 /**
   2  * collectd - src/email.c
   3  * Copyright (C) 2006  Sebastian Harl
   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; either version 2 of the License, or (at your
   8  * option) any later version.
   9  *
  10  * This program 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  * General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
  18  *
  19  * Author:
  20  *   Sebastian Harl <sh at tokkee.org>
  21  **/
  22
  23 /*
  24  * This plugin communicates with a spam filter, a virus scanner or similar
  25  * software using a UNIX socket and a very simple protocol:
  26  *
  27  * e-mail type (e.g. ham, spam, virus, ...) and size
  28  * e:<type>:<bytes>
  29  *
  30  * spam score
  31  * s:<value>
  32  *
  33  * successful spam checks
  34  * c:<type1>[,<type2>,...]
  35  */
  36
  37 #include "collectd.h"
  38 #include "common.h"
  39 #include "plugin.h"
  40
  41 #if HAVE_LIBPTHREAD
  42 # include <pthread.h>
  43 # define EMAIL_HAVE_READ 1
  44 #else
  45 # define EMAIL_HAVE_READ 0
  46 #endif
  47
  48 #if HAVE_SYS_SELECT_H
  49 #       include <sys/select.h>
  50 #endif /* HAVE_SYS_SELECT_H */
  51
  52 #if HAVE_SYS_SOCKET_H
  53 #       include <sys/socket.h>
  54 #endif /* HAVE_SYS_SOCKET_H */
  55
  56 /* *sigh* glibc does not define UNIX_PATH_MAX in sys/un.h ... */
  57 #if HAVE_LINUX_UN_H
  58 #       include <linux/un.h>
  59 #elif HAVE_SYS_UN_H
  60 #       include <sys/un.h>
  61 #endif /* HAVE_LINUX_UN_H | HAVE_SYS_UN_H */
  62
  63 #if HAVE_GRP_H
  64 #       include <grp.h>
  65 #endif /* HAVE_GRP_H */
  66
  67 #define MODULE_NAME "email"
  68
  69 /* 256 bytes ought to be enough for anybody ;-) */
  70 #define BUFSIZE 256
  71
  72 #define SOCK_PATH "/tmp/.collectd-email"
  73 #define MAX_CONNS 5
  74
  75 /*
  76  * Private data structures
  77  */
  78 #if EMAIL_HAVE_READ
  79 /* linked list of email and check types */
  80 typedef struct type {
  81         char        *name;
  82         int         value;
  83         struct type *next;
  84 } type_t;
  85
  86 typedef struct {
  87         type_t *head;
  88         type_t *tail;
  89 } type_list_t;
  90
  91 /* linked list of collector thread control information */
  92 typedef struct collector {
  93         pthread_t thread;
  94
  95         /* socket to read data from */
  96         int socket;
  97
  98         /* buffer to read data to */
  99         char buffer[BUFSIZE];
 100         int  idx; /* current position in buffer */
 101
 102         struct collector *next;
 103 } collector_t;
 104
 105 typedef struct {
 106         collector_t *head;
 107         collector_t *tail;
 108 } collector_list_t;
 109 #endif /* EMAIL_HAVE_READ */
 110
 111 /*
 112  * Private variables
 113  */
 114 #if EMAIL_HAVE_READ
 115 /* state of the plugin */
 116 static int disabled = 0;
 117
 118 /* thread managing "client" connections */
 119 static pthread_t connector;
 120
 121 /* tell the connector thread that a collector is available */
 122 static pthread_cond_t collector_available = PTHREAD_COND_INITIALIZER;
 123
 124 /* collector threads that are in use */
 125 static pthread_mutex_t active_mutex = PTHREAD_MUTEX_INITIALIZER;
 126 static collector_list_t active;
 127
 128 /* collector threads that are available for use */
 129 static pthread_mutex_t available_mutex = PTHREAD_MUTEX_INITIALIZER;
 130 static collector_list_t available;
 131
 132 static pthread_mutex_t count_mutex = PTHREAD_MUTEX_INITIALIZER;
 133 static type_list_t count;
 134
 135 static pthread_mutex_t size_mutex = PTHREAD_MUTEX_INITIALIZER;
 136 static type_list_t size;
 137
 138 static pthread_mutex_t score_mutex = PTHREAD_MUTEX_INITIALIZER;
 139 static double score;
 140 static int score_count;
 141
 142 static pthread_mutex_t check_mutex = PTHREAD_MUTEX_INITIALIZER;
 143 static type_list_t check;
 144 #endif /* EMAIL_HAVE_READ */
 145
 146 #define COUNT_FILE "email/email-%s.rrd"
 147 static char *count_ds_def[] =
 148 {
 149         "DS:count:GAUGE:"COLLECTD_HEARTBEAT":0:U",
 150         NULL
 151 };
 152 static int count_ds_num = 1;
 153
 154 #define SIZE_FILE  "email/email_size-%s.rrd"
 155 static char *size_ds_def[] =
 156 {
 157         "DS:size:GAUGE:"COLLECTD_HEARTBEAT":0:U",
 158         NULL
 159 };
 160 static int size_ds_num = 1;
 161
 162 #define SCORE_FILE "email/spam_score.rrd"
 163 static char *score_ds_def[] =
 164 {
 165         "DS:score:GAUGE:"COLLECTD_HEARTBEAT":U:U",
 166         NULL
 167 };
 168 static int score_ds_num = 1;
 169
 170 #define CHECK_FILE "email/spam_check-%s.rrd"
 171 static char *check_ds_def[] =
 172 {
 173         "DS:hits:GAUGE:"COLLECTD_HEARTBEAT":0:U",
 174         NULL
 175 };
 176 static int check_ds_num = 1;
 177
 178 #if EMAIL_HAVE_READ
 179 /* Increment the value of the given name in the given list by incr. */
 180 static void type_list_incr (type_list_t *list, char *name, int incr)
 181 {
 182         if (NULL == list->head) {
 183                 list->head = (type_t *)smalloc (sizeof (type_t));
 184
 185                 list->head->name  = sstrdup (name);
 186                 list->head->value = incr;
 187                 list->head->next  = NULL;
 188
 189                 list->tail = list->head;
 190         }
 191         else {
 192                 type_t *ptr;
 193
 194                 for (ptr = list->head; NULL != ptr; ptr = ptr->next) {
 195                         if (0 == strcmp (name, ptr->name))
 196                                 break;
 197                 }
 198
 199                 if (NULL == ptr) {
 200                         list->tail->next = (type_t *)smalloc (sizeof (type_t));
 201                         list->tail = list->tail->next;
 202
 203                         list->tail->name  = sstrdup (name);
 204                         list->tail->value = incr;
 205                         list->tail->next  = NULL;
 206                 }
 207                 else {
 208                         ptr->value += incr;
 209                 }
 210         }
 211         return;
 212 } /* static void type_list_incr (type_list_t *, char *) */
 213
 214 /* Read a single character from the socket. If an error occurs or end-of-file
 215  * is reached return '\0'. */
 216 char read_char (collector_t *src)
 217 {
 218         char ret = '\0';
 219
 220         fd_set fdset;
 221
 222         FD_ZERO (&fdset);
 223         FD_SET (src->socket, &fdset);
 224
 225         if (-1 == select (src->socket + 1, &fdset, NULL, NULL, NULL)) {
 226                 syslog (LOG_ERR, "select() failed: %s", strerror (errno));
 227                 return '\0';
 228         }
 229
 230         assert (FD_ISSET (src->socket, &fdset));
 231
 232         do {
 233                 ssize_t len = 0;
 234
 235                 errno = 0;
 236                 if (0 > (len = read (src->socket, (void *)&ret, 1))) {
 237                         if (EINTR != errno) {
 238                                 syslog (LOG_ERR, "read() failed: %s", strerror (errno));
 239                                 return '\0';
 240                         }
 241                 }
 242
 243                 if (0 == len)
 244                         return '\0';
 245         } while (EINTR == errno);
 246         return ret;
 247 } /* char read_char (collector_t *) */
 248
 249 /* Read a single line (terminated by '\n') from the the socket.
 250  *
 251  * The return value is zero terminated and does not contain any newline
 252  * characters. In case that no complete line is available (non-blocking mode
 253  * should be enabled) an empty string is returned.
 254  *
 255  * If an error occurs or end-of-file is reached return NULL.
 256  *
 257  * IMPORTANT NOTE: If there is no newline character found in BUFSIZE
 258  * characters of the input stream, the line will will be ignored! By
 259  * definition we should not get any longer input lines, thus this is
 260  * acceptable in this case ;-) */
 261 char *read_line (collector_t *src)
 262 {
 263         int  i = 0;
 264         char *ret;
 265
 266         assert (BUFSIZE > src->idx);
 267
 268         for (i = 0; i < src->idx; ++i) {
 269                 if ('\n' == src->buffer[i])
 270                         break;
 271         }
 272
 273         if ('\n' != src->buffer[i]) {
 274                 fd_set fdset;
 275
 276                 ssize_t len = 0;
 277
 278                 FD_ZERO (&fdset);
 279                 FD_SET (src->socket, &fdset);
 280
 281                 if (-1 == select (src->socket + 1, &fdset, NULL, NULL, NULL)) {
 282                         syslog (LOG_ERR, "select() failed: %s", strerror (errno));
 283                         return NULL;
 284                 }
 285
 286                 assert (FD_ISSET (src->socket, &fdset));
 287
 288                 do {
 289                         errno = 0;
 290                         if (0 > (len = read (src->socket,
 291                                                         (void *)(&(src->buffer[0]) + src->idx),
 292                                                         BUFSIZE - src->idx))) {
 293                                 if (EINTR != errno) {
 294                                         syslog (LOG_ERR, "read() failed: %s", strerror (errno));
 295                                         return NULL;
 296                                 }
 297                         }
 298
 299                         if (0 == len)
 300                                 return NULL;
 301                 } while (EINTR == errno);
 302
 303                 src->idx += len;
 304
 305                 for (i = src->idx - len; i < src->idx; ++i) {
 306                         if ('\n' == src->buffer[i])
 307                                 break;
 308                 }
 309
 310                 if ('\n' != src->buffer[i]) {
 311                         ret = (char *)smalloc (1);
 312
 313                         ret[0] = '\0';
 314
 315                         if (BUFSIZE == src->idx) { /* no space left in buffer */
 316                                 while ('\n' != read_char (src))
 317                                         /* ignore complete line */;
 318
 319                                 src->idx = 0;
 320                         }
 321                         return ret;
 322                 }
 323         }
 324
 325         ret = (char *)smalloc (i + 1);
 326         memcpy (ret, &(src->buffer[0]), i + 1);
 327         ret[i] = '\0';
 328
 329         src->idx -= (i + 1);
 330
 331         if (0 == src->idx)
 332                 src->buffer[0] = '\0';
 333         else
 334                 memmove (&(src->buffer[0]), &(src->buffer[i + 1]), src->idx);
 335         return ret;
 336 } /* char *read_line (collector_t *) */
 337
 338 static void *collect (void *arg)
 339 {
 340         collector_t *this = (collector_t *)arg;
 341
 342         int loop = 1;
 343
 344         { /* put the socket in non-blocking mode */
 345                 int flags = 0;
 346
 347                 errno = 0;
 348                 if (-1 == fcntl (this->socket, F_GETFL, &flags)) {
 349                         syslog (LOG_ERR, "fcntl() failed: %s", strerror (errno));
 350                         loop = 0;
 351                 }
 352
 353                 errno = 0;
 354                 if (-1 == fcntl (this->socket, F_SETFL, flags | O_NONBLOCK)) {
 355                         syslog (LOG_ERR, "fcntl() failed: %s", strerror (errno));
 356                         loop = 0;
 357                 }
 358         }
 359
 360         while (loop) {
 361                 char *line = read_line (this);
 362
 363                 if (NULL == line) {
 364                         loop = 0;
 365                         break;
 366                 }
 367
 368                 if ('\0' == line[0]) {
 369                         free (line);
 370                         continue;
 371                 }
 372
 373                 if (':' != line[1]) {
 374                         syslog (LOG_ERR, "email: syntax error in line '%s'", line);
 375                         free (line);
 376                         continue;
 377                 }
 378
 379                 if ('e' == line[0]) { /* e:<type>:<bytes> */
 380                         char *ptr  = NULL;
 381                         char *type = strtok_r (line + 2, ":", &ptr);
 382                         char *tmp  = strtok_r (NULL, ":", &ptr);
 383                         int  bytes = 0;
 384
 385                         if (NULL == tmp) {
 386                                 syslog (LOG_ERR, "email: syntax error in line '%s'", line);
 387                                 free (line);
 388                                 continue;
 389                         }
 390
 391                         bytes = atoi (tmp);
 392
 393                         pthread_mutex_lock (&count_mutex);
 394                         type_list_incr (&count, type, 1);
 395                         pthread_mutex_unlock (&count_mutex);
 396
 397                         pthread_mutex_lock (&size_mutex);
 398                         type_list_incr (&size, type, bytes);
 399                         pthread_mutex_unlock (&size_mutex);
 400                 }
 401                 else if ('s' == line[0]) { /* s:<value> */
 402                         pthread_mutex_lock (&score_mutex);
 403                         score = (score * (double)score_count + atof (line + 2))
 404                                         / (double)(score_count + 1);
 405                         ++score_count;
 406                         pthread_mutex_unlock (&score_mutex);
 407                 }
 408                 else if ('c' == line[0]) { /* c:<type1>[,<type2>,...] */
 409                         char *ptr  = NULL;
 410                         char *type = strtok_r (line + 2, ",", &ptr);
 411
 412                         do {
 413                                 pthread_mutex_lock (&check_mutex);
 414                                 type_list_incr (&check, type, 1);
 415                                 pthread_mutex_unlock (&check_mutex);
 416                         } while (NULL != (type = strtok_r (NULL, ",", &ptr)));
 417                 }
 418                 else {
 419                         syslog (LOG_ERR, "email: unknown type '%c'", line[0]);
 420                 }
 421
 422                 free (line);
 423         }
 424
 425         /* put this thread back into the available list */
 426         pthread_mutex_lock (&active_mutex);
 427         {
 428                 collector_t *last;
 429                 collector_t *ptr;
 430
 431                 last = NULL;
 432
 433                 for (ptr = active.head; NULL != ptr; last = ptr, ptr = ptr->next) {
 434                         if (0 != pthread_equal (ptr->thread, this->thread))
 435                                 break;
 436                 }
 437
 438                 /* the current thread _has_ to be in the active list */
 439                 assert (NULL != ptr);
 440
 441                 if (NULL == last) {
 442                         active.head = ptr->next;
 443                 }
 444                 else {
 445                         last->next = ptr->next;
 446
 447                         if (NULL == last->next) {
 448                                 active.tail = last;
 449                         }
 450                 }
 451         }
 452         pthread_mutex_unlock (&active_mutex);
 453
 454         this->next = NULL;
 455
 456         pthread_mutex_lock (&available_mutex);
 457
 458         if (NULL == available.head) {
 459                 available.head = this;
 460                 available.tail = this;
 461         }
 462         else {
 463                 available.tail->next = this;
 464                 available.tail = this;
 465         }
 466
 467         pthread_mutex_unlock (&available_mutex);
 468
 469         pthread_cond_signal (&collector_available);
 470         pthread_exit ((void *)0);
 471 } /* void *collect (void *) */
 472
 473 static void *open_connection (void *arg)
 474 {
 475         int local = 0;
 476
 477         struct sockaddr_un addr;
 478
 479         /* create UNIX socket */
 480         errno = 0;
 481         if (-1 == (local = socket (PF_UNIX, SOCK_STREAM, 0))) {
 482                 disabled = 1;
 483                 syslog (LOG_ERR, "socket() failed: %s", strerror (errno));
 484                 pthread_exit ((void *)1);
 485         }
 486
 487         addr.sun_family = AF_UNIX;
 488
 489         strncpy (addr.sun_path, SOCK_PATH, (size_t)(UNIX_PATH_MAX - 1));
 490         addr.sun_path[UNIX_PATH_MAX - 1] = '\0';
 491         unlink (addr.sun_path);
 492
 493         errno = 0;
 494         if (-1 == bind (local, (struct sockaddr *)&addr,
 495                                 offsetof (struct sockaddr_un, sun_path)
 496                                         + strlen(addr.sun_path))) {
 497                 disabled = 1;
 498                 syslog (LOG_ERR, "bind() failed: %s", strerror (errno));
 499                 pthread_exit ((void *)1);
 500         }
 501
 502         errno = 0;
 503         if (-1 == listen (local, 5)) {
 504                 disabled = 1;
 505                 syslog (LOG_ERR, "listen() failed: %s", strerror (errno));
 506                 pthread_exit ((void *)1);
 507         }
 508
 509         if ((uid_t)0 == geteuid ()) {
 510                 struct group *grp;
 511
 512                 errno = 0;
 513                 if (NULL != (grp = getgrnam (COLLECTD_GRP_NAME))) {
 514                         errno = 0;
 515                         if (0 != chown (SOCK_PATH, (uid_t)-1, grp->gr_gid)) {
 516                                 syslog (LOG_WARNING, "chown() failed: %s", strerror (errno));
 517                         }
 518                 }
 519                 else {
 520                         syslog (LOG_WARNING, "getgrnam() failed: %s", strerror (errno));
 521                 }
 522         }
 523         else {
 524                 syslog (LOG_WARNING, "not running as root");
 525         }
 526
 527         errno = 0;
 528         if (0 != chmod (SOCK_PATH, S_IRWXU | S_IRWXG)) {
 529                 syslog (LOG_WARNING, "chmod() failed: %s", strerror (errno));
 530         }
 531
 532         { /* initialize queue of available threads */
 533                 int i = 0;
 534
 535                 collector_t *last;
 536
 537                 active.head = NULL;
 538                 active.tail = NULL;
 539
 540                 available.head = (collector_t *)smalloc (sizeof (collector_t));
 541                 available.tail = available.head;
 542                 available.tail->next = NULL;
 543
 544                 last = available.head;
 545
 546                 for (i = 1; i < MAX_CONNS; ++i) {
 547                         last->next = (collector_t *)smalloc (sizeof (collector_t));
 548                         last = last->next;
 549                         available.tail = last;
 550                         available.tail->next = NULL;
 551                 }
 552         }
 553
 554         while (1) {
 555                 int remote = 0;
 556                 int err    = 0;
 557
 558                 collector_t *collector;
 559
 560                 pthread_attr_t ptattr;
 561
 562                 pthread_mutex_lock (&available_mutex);
 563                 while (NULL == available.head) {
 564                         pthread_cond_wait (&collector_available, &available_mutex);
 565                 }
 566                 pthread_mutex_unlock (&available_mutex);
 567
 568                 do {
 569                         errno = 0;
 570                         if (-1 == (remote = accept (local, NULL, NULL))) {
 571                                 if (EINTR != errno) {
 572                                         disabled = 1;
 573                                         syslog (LOG_ERR, "accept() failed: %s", strerror (errno));
 574                                         pthread_exit ((void *)1);
 575                                 }
 576                         }
 577                 } while (EINTR == errno);
 578
 579                 /* assign connection to next available thread */
 580                 pthread_mutex_lock (&available_mutex);
 581
 582                 collector = available.head;
 583                 collector->socket = remote;
 584
 585                 if (available.head == available.tail) {
 586                         available.head = NULL;
 587                         available.tail = NULL;
 588                 }
 589                 else {
 590                         available.head = available.head->next;
 591                 }
 592
 593                 pthread_mutex_unlock (&available_mutex);
 594
 595                 collector->idx  = 0;
 596                 collector->next = NULL;
 597
 598                 pthread_attr_init (&ptattr);
 599                 pthread_attr_setdetachstate (&ptattr, PTHREAD_CREATE_DETACHED);
 600
 601                 if (0 == (err = pthread_create (&collector->thread, &ptattr, collect,
 602                                 (void *)collector))) {
 603                         pthread_mutex_lock (&active_mutex);
 604
 605                         if (NULL == active.head) {
 606                                 active.head = collector;
 607                                 active.tail = collector;
 608                         }
 609                         else {
 610                                 active.tail->next = collector;
 611                                 active.tail = collector;
 612                         }
 613
 614                         pthread_mutex_unlock (&active_mutex);
 615                 }
 616                 else {
 617                         pthread_mutex_lock (&available_mutex);
 618
 619                         if (NULL == available.head) {
 620                                 available.head = collector;
 621                                 available.tail = collector;
 622                         }
 623                         else {
 624                                 available.tail->next = collector;
 625                                 available.tail = collector;
 626                         }
 627
 628                         pthread_mutex_unlock (&available_mutex);
 629
 630                         close (remote);
 631                         syslog (LOG_ERR, "pthread_create() failed: %s", strerror (err));
 632                 }
 633
 634                 pthread_attr_destroy (&ptattr);
 635         }
 636         pthread_exit ((void *)0);
 637 } /* void *open_connection (void *) */
 638 #endif /* EMAIL_HAVE_READ */
 639
 640 static void email_init (void)
 641 {
 642 #if EMAIL_HAVE_READ
 643         int err = 0;
 644
 645         if (0 != (err = pthread_create (&connector, NULL,
 646                                 open_connection, NULL))) {
 647                 disabled = 1;
 648                 syslog (LOG_ERR, "pthread_create() failed: %s", strerror (err));
 649                 return;
 650         }
 651 #endif /* EMAIL_HAVE_READ */
 652         return;
 653 } /* static void email_init (void) */
 654
 655 static void count_write (char *host, char *inst, char *val)
 656 {
 657         char file[BUFSIZE] = "";
 658         int  len           = 0;
 659
 660         len = snprintf (file, BUFSIZE, COUNT_FILE, inst);
 661         if ((len < 0) || (len >= BUFSIZE))
 662                 return;
 663
 664         rrd_update_file (host, file, val, count_ds_def, count_ds_num);
 665         return;
 666 } /* static void email_write (char *host, char *inst, char *val) */
 667
 668 static void size_write (char *host, char *inst, char *val)
 669 {
 670         char file[BUFSIZE] = "";
 671         int  len           = 0;
 672
 673         len = snprintf (file, BUFSIZE, SIZE_FILE, inst);
 674         if ((len < 0) || (len >= BUFSIZE))
 675                 return;
 676
 677         rrd_update_file (host, file, val, size_ds_def, size_ds_num);
 678         return;
 679 } /* static void size_write (char *host, char *inst, char *val) */
 680
 681 static void score_write (char *host, char *inst, char *val)
 682 {
 683         rrd_update_file (host, SCORE_FILE, val, score_ds_def, score_ds_num);
 684         return;
 685 } /* static void score_write (char *host, char *inst, char *val) */
 686
 687 static void check_write (char *host, char *inst, char *val)
 688 {
 689         char file[BUFSIZE] = "";
 690         int  len           = 0;
 691
 692         len = snprintf (file, BUFSIZE, CHECK_FILE, inst);
 693         if ((len < 0) || (len >= BUFSIZE))
 694                 return;
 695
 696         rrd_update_file (host, file, val, check_ds_def, check_ds_num);
 697         return;
 698 } /* static void check_write (char *host, char *inst, char *val) */
 699
 700 #if EMAIL_HAVE_READ
 701 static void type_submit (char *plugin, char *inst, int value)
 702 {
 703         char buf[BUFSIZE] = "";
 704         int  len          = 0;
 705
 706         if (0 == value)
 707                 return;
 708
 709         len = snprintf (buf, BUFSIZE, "%u:%i", (unsigned int)curtime, value);
 710         if ((len < 0) || (len >= BUFSIZE))
 711                 return;
 712
 713         plugin_submit (plugin, inst, buf);
 714         return;
 715 } /* static void type_submit (char *, char *, int) */
 716
 717 static void score_submit (double value)
 718 {
 719         char buf[BUFSIZE] = "";
 720         int  len          = 0;
 721
 722         if (0.0 == value)
 723                 return;
 724
 725         len = snprintf (buf, BUFSIZE, "%u:%.2f", (unsigned int)curtime, value);
 726         if ((len < 0) || (len >= BUFSIZE))
 727                 return;
 728
 729         plugin_submit ("email_spam_score", NULL, buf);
 730         return;
 731 }
 732
 733 static void email_read (void)
 734 {
 735         type_t *ptr;
 736
 737         if (disabled)
 738                 return;
 739
 740         pthread_mutex_lock (&count_mutex);
 741
 742         for (ptr = count.head; NULL != ptr; ptr = ptr->next) {
 743                 type_submit ("email_count", ptr->name, ptr->value);
 744                 ptr->value = 0;
 745         }
 746
 747         pthread_mutex_unlock (&count_mutex);
 748
 749         pthread_mutex_lock (&size_mutex);
 750
 751         for (ptr = size.head; NULL != ptr; ptr = ptr->next) {
 752                 type_submit ("email_size", ptr->name, ptr->value);
 753                 ptr->value = 0;
 754         }
 755
 756         pthread_mutex_unlock (&size_mutex);
 757
 758         pthread_mutex_lock (&score_mutex);
 759
 760         score_submit (score);
 761         score = 0.0;
 762         score_count = 0;
 763
 764         pthread_mutex_unlock (&score_mutex);
 765
 766         pthread_mutex_lock (&check_mutex);
 767
 768         for (ptr = check.head; NULL != ptr; ptr = ptr->next) {
 769                 type_submit ("email_spam_check", ptr->name, ptr->value);
 770                 ptr->value = 0;
 771         }
 772
 773         pthread_mutex_unlock (&check_mutex);
 774         return;
 775 } /* static void read (void) */
 776 #else /* if !EMAIL_HAVE_READ */
 777 # define email_read NULL
 778 #endif
 779
 780 void module_register (void)
 781 {
 782         plugin_register (MODULE_NAME, email_init, email_read, NULL);
 783         plugin_register ("email_count", NULL, NULL, count_write);
 784         plugin_register ("email_size", NULL, NULL, size_write);
 785         plugin_register ("email_spam_score", NULL, NULL, score_write);
 786         plugin_register ("email_spam_check", NULL, NULL, check_write);
 787         return;
 788 } /* void module_register (void) */
 789
 790 /* vim: set sw=4 ts=4 tw=78 noexpandtab : */
 791