src/xport_to_queue.c

   1 /*======================================================================*
   2  * Copyright (c) 2008, Yahoo! Inc. All rights reserved.                 *
   3  *                                                                      *
   4  * Licensed under the New BSD License (the "License"); you may not use  *
   5  * this file except in compliance with the License.  Unless required    *
   6  * by applicable law or agreed to in writing, software distributed      *
   7  * under the License is distributed on an "AS IS" BASIS, WITHOUT        *
   8  * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.     *
   9  * See the License for the specific language governing permissions and  *
  10  * limitations under the License. See accompanying LICENSE file.        *
  11  *======================================================================*/
  12
  13 #include "config.h"
  14
  15 #include "xport_to_queue.h"
  16
  17 #include "header.h"
  18 #include "opt.h"
  19 #include "perror.h"
  20 #include "queue.h"
  21 #include "sig.h"
  22 #include "xport.h"
  23 #include "stats.h"
  24
  25 #include <fcntl.h>
  26 #include <stdio.h>
  27 #include <stdlib.h>
  28 #include <unistd.h>
  29
  30 static void skd(void);
  31
  32 struct enqueuer_stats est ;
  33
  34 void* xport_to_queue(void* arg)
  35 {
  36   struct xport xpt;
  37   struct queue que;
  38
  39   unsigned char* buf = 0;
  40   size_t bufsiz;
  41
  42   (void)arg; /* appease -Wall -Werror */
  43
  44   install_signal_handlers();
  45
  46   if ( arg_rt )
  47     {
  48       skd();
  49     }
  50
  51   if ( (queue_factory(&que) < 0) || (que.vtbl->open(&que, O_WRONLY) < 0) )
  52     {
  53       LOG_ER("Failed to create or open queue object.\n");
  54       exit(EXIT_FAILURE);
  55     }
  56
  57   /* Can we drop root here? */
  58
  59   if ( (xport_factory(&xpt) < 0) || (xpt.vtbl->open(&xpt, O_RDONLY) < 0) )
  60     {
  61       LOG_ER("Failed to create xport object.\n");
  62       exit(EXIT_FAILURE);
  63     }
  64
  65   buf = (unsigned char*)que.vtbl->alloc(&que, &bufsiz);
  66   if ( 0 == buf )
  67     {
  68       LOG_ER("unable to allocate %d bytes for message buffer.\n", bufsiz);
  69       exit(EXIT_FAILURE);
  70     }
  71   memset(buf, 0, HEADER_LENGTH); /* Clear the header portion of the message. */
  72
  73   /* Read a packet from the transport, write it to the queue. */
  74   while ( ! gbl_done )
  75     {
  76       int xpt_read_ret;
  77       int que_write_ret;
  78
  79       unsigned long long tm;
  80       unsigned long addr;
  81       short port;
  82
  83       if ( (xpt_read_ret = xpt.vtbl->read(&xpt,
  84                                           buf + HEADER_LENGTH,
  85                                           bufsiz - HEADER_LENGTH,
  86                                           &addr, &port)) < 0 )
  87         {
  88           enqueuer_stats_record_socket_error(&est);
  89           continue;
  90         }
  91       tm = time_in_milliseconds();
  92
  93       enqueuer_stats_record_datagram(&est,xpt_read_ret);
  94
  95       /* Return info about packet read. */
  96       LOG_PROG("Read %d bytes\n", xpt_read_ret);
  97       LOG_PROG("From %08lx:%04x.\n", addr, port&0xffff);
  98
  99       header_add(buf, xpt_read_ret, tm, addr, port);
 100
 101       if ( header_is_rotate (buf) )
 102         { // Command::Rotate: here we just collect some stats.
 103           enqueuer_stats_rotate(&est);
 104         }
 105
 106       if ( (que_write_ret = que.vtbl->write(&que,
 107                                             buf,
 108                                             xpt_read_ret + HEADER_LENGTH)) < 0 )
 109         {
 110           LOG_ER("Queue write error attempting to write %d bytes.\n",
 111                  xpt_read_ret + HEADER_LENGTH);
 112           continue;
 113         }
 114       else
 115         {
 116           LOG_PROG("Queue write of %d bytes.\n",
 117                    xpt_read_ret + HEADER_LENGTH);
 118         }
 119     }
 120   que.vtbl->dealloc(&que, buf);
 121
 122   xpt.vtbl->destructor(&xpt);
 123   que.vtbl->destructor(&que);
 124
 125   enqueuer_stats_report(&est);
 126
 127   return 0;
 128 }
 129
 130 #if HAVE_SCHED_H
 131 #include <sched.h>
 132
 133 static void skd()
 134 {
 135   struct sched_param sp;
 136
 137   sp.sched_priority = sched_get_priority_max(SCHED_FIFO);
 138   if ( sched_setscheduler(0, SCHED_FIFO, &sp) )
 139     {
 140       PERROR("sched_setscheduler");
 141       LOG_WARN("Increasing thread priority failed"
 142                ", will run with standard priorities");
 143     }
 144   else
 145     {
 146       LOG_WARN("Running with FIFO priority");
 147     }
 148 }
 149 #else
 150 static void skd()
 151 {
 152   LOG_WARN("No real-time scheduler support on this platform.\n");
 153 }
 154 #endif