release/src/router/udpxy/ctx.c

   1 /* @(#) client/server context implementation
   2  *
   3  * Copyright 2008-2011 Pavel V. Cherenkov (pcherenkov@gmail.com)
   4  *
   5  *  This file is part of udpxy.
   6  *
   7  *  udpxy is free software: you can redistribute it and/or modify
   8  *  it under the terms of the GNU General Public License as published by
   9  *  the Free Software Foundation, either version 3 of the License, or
  10  *  (at your option) any later version.
  11  *
  12  *  udpxy is distributed in the hope that it will be useful,
  13  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  *  GNU General Public License for more details.
  16  *
  17  *  You should have received a copy of the GNU General Public License
  18  *  along with udpxy.  If not, see <http://www.gnu.org/licenses/>.
  19  */
  20
  21 #include "osdef.h"  /* os-specific definitions */
  22
  23 #include <sys/types.h>
  24 #include <sys/socket.h>
  25 #include <sys/stat.h>
  26 #include <arpa/inet.h>
  27 #include <net/if.h>
  28 #include <netinet/in.h>
  29
  30 #include <string.h>
  31 #include <assert.h>
  32 #include <stdlib.h>
  33 #include <stdio.h>
  34 #include <errno.h>
  35 #include <fcntl.h>
  36 #include <unistd.h>
  37 #include <time.h>
  38
  39 #include "ctx.h"
  40 #include "udpxy.h"
  41 #include "util.h"
  42 #include "mtrace.h"
  43
  44 extern FILE* g_flog;
  45 extern const char IPv4_ALL[];
  46
  47 /* initialize server context data
  48  */
  49 int
  50 init_server_ctx( struct server_ctx* ctx,
  51                  const size_t max,
  52                  const char* laddr, uint16_t lport,
  53                  const char* mifc_addr )
  54 {
  55     int flags = -1;
  56
  57     assert( lport && mifc_addr && ctx && max );
  58
  59     ctx->lsockfd = 0;
  60     (void) strncpy( ctx->listen_addr, (laddr ? laddr : IPv4_ALL),
  61                     IPADDR_STR_SIZE );
  62     ctx->listen_addr[ IPADDR_STR_SIZE - 1 ] = '\0';
  63
  64     ctx->listen_port = lport;
  65
  66     (void) strncpy( ctx->mcast_ifc_addr, mifc_addr, IPADDR_STR_SIZE );
  67     ctx->mcast_ifc_addr[ IPADDR_STR_SIZE - 1 ] = '\0';
  68
  69     ctx->cl = calloc(max, sizeof(struct client_ctx));
  70     if( NULL == ctx->cl ) {
  71         mperror( g_flog, errno, "%s: client_t - calloc", __func__ );
  72         return ERR_INTERNAL;
  73     }
  74
  75     (void) memset( ctx->cl, 0, max * sizeof(struct client_ctx) );
  76     ctx->clfree = ctx->clmax = max;
  77
  78     if( 0 != pipe(ctx->cpipe) ) {
  79         mperror( g_flog, errno, "%s: pipe", __func__ );
  80         return ERR_INTERNAL;
  81     }
  82
  83     /* make reading end of pipe non-blocking (we don't want to
  84      * block on pipe read on the server side)
  85      */
  86     if( -1 == (flags = fcntl( ctx->cpipe[0], F_GETFL )) ||
  87         -1 == fcntl( ctx->cpipe[0], F_SETFL, flags | O_NONBLOCK ) ) {
  88         mperror( g_flog, errno, "%s: fcntl", __func__ );
  89         return ERR_INTERNAL;
  90     }
  91
  92     return 0;
  93 }
  94
  95
  96 void
  97 free_server_ctx( struct server_ctx* ctx )
  98 {
  99     int i = -1;
 100
 101     assert(ctx);
 102     free( ctx->cl );
 103
 104     for( i = 0; i < 2; ++i ) {
 105         if( ctx->cpipe[i] < 0 ) continue;
 106
 107         if( 0 != close( ctx->cpipe[i] ) ) {
 108             mperror( g_flog, errno,
 109                     "%s: close(pipe)", __func__ );
 110         }
 111     }
 112
 113     (void) memset( ctx, 0, sizeof(struct server_ctx) );
 114     return;
 115 }
 116
 117
 118 /* find index of the first client with the given pid
 119  */
 120 int
 121 find_client( const struct server_ctx* ctx, pid_t pid )
 122 {
 123     int i = -1;
 124
 125     assert( ctx && (pid >= 0) );
 126
 127     for( i = 0; (size_t)i < ctx->clmax; ++i ) {
 128         if( ctx->cl[ i ].pid == pid )
 129             return i;
 130     }
 131
 132     return -1;
 133 }
 134
 135 /* populate connection's source info in client context
 136  */
 137 static int
 138 get_src_info( struct client_ctx* cl, int sockfd )
 139 {
 140     struct stat st;
 141     struct sockaddr_in addr;
 142     a_socklen_t len = 0;
 143     int rc = 0;
 144
 145     assert( cl );
 146     if( -1 == (rc = fstat( sockfd, &st )) ) {
 147         mperror( g_flog, errno, "%s: fstat", __func__ );
 148         return ERR_INTERNAL;
 149     }
 150
 151     if( S_ISREG( st.st_mode ) ) {
 152         (void) strncpy( cl->src_addr, "File", sizeof(cl->src_addr) );
 153         cl->src_addr[ sizeof(cl->src_addr) - 1 ] = '\0';
 154         cl->src_port = 0;
 155     }
 156     else if( S_ISSOCK( st.st_mode ) ) {
 157         len = sizeof(addr);
 158         rc = getpeername( sockfd, (struct sockaddr*)&addr, &len );
 159         if( 0 == rc ) {
 160             (void)strncpy( cl->src_addr, inet_ntoa(addr.sin_addr),
 161                     sizeof(cl->src_addr) - 1 );
 162             cl->src_addr[ sizeof(cl->src_addr) - 1 ] = '\0';
 163
 164             cl->src_port = ntohs(addr.sin_port);
 165         }
 166         else {
 167             mperror( g_flog, errno, "%s: getpeername", __func__ );
 168             return ERR_INTERNAL;
 169         }
 170     } /* S_ISSOCK */
 171
 172     return rc;
 173 }
 174
 175
 176 /* add client to server context
 177  */
 178 int
 179 add_client( struct server_ctx* ctx,
 180             pid_t cpid, const char* maddr, uint16_t mport,
 181             int sockfd )
 182 {
 183     struct client_ctx* client = NULL;
 184     int index = -1;
 185     int rc = 0;
 186
 187     assert( ctx && maddr && mport );
 188
 189     index = find_client( ctx, 0 );
 190     if( -1 == index )
 191         return ERR_INTERNAL;
 192
 193     client = &(ctx->cl[ index ]);
 194     client->pid = cpid;
 195
 196     (void) strncpy( client->mcast_addr, maddr, IPADDR_STR_SIZE );
 197     client->mcast_addr[ IPADDR_STR_SIZE - 1 ] = '\0';
 198
 199     client->mcast_port = mport;
 200
 201     rc = get_src_info( client, sockfd );
 202     if( 0 != rc ) {
 203         client->pid = 0;
 204         return ERR_INTERNAL;
 205     }
 206
 207     /* init sender id: 0 indicates no stats */
 208     client->tstat.sender_id = 0;
 209
 210     ctx->clfree--;
 211
 212     if( g_flog ) {
 213         (void)tmfprintf( g_flog, "Added client: pid=[%d], maddr=[%s], mport=[%d], "
 214                 "saddr=[%s], sport=[%d]\n",
 215                 (int)cpid, maddr, (int)mport, client->src_addr, client->src_port );
 216     }
 217     return 0;
 218 }
 219
 220
 221 /* delete client from server context
 222  */
 223 int
 224 delete_client( struct server_ctx* ctx, pid_t cpid )
 225 {
 226     struct client_ctx* client = NULL;
 227     int index = -1;
 228
 229     assert( ctx && (cpid > 0) );
 230
 231     index = find_client( ctx, cpid );
 232     if( -1 == index ) {
 233         return ERR_INTERNAL;
 234     }
 235
 236     client = &(ctx->cl[ index ]);
 237     client->pid = 0;
 238
 239     ctx->clfree++;
 240
 241     if( g_flog ) {
 242         TRACE( (void)tmfprintf( g_flog, "Deleted client: pid=[%d]\n", cpid) );
 243     }
 244
 245     return 0;
 246 }
 247
 248
 249 /* init traffic relay statistics
 250  */
 251 void
 252 tpstat_init( struct tps_data* d, int setpid )
 253 {
 254     assert( d );
 255
 256     if( setpid )
 257         d->pid = getpid();
 258
 259     d->tm_from = time(NULL);
 260     d->niter = 0;
 261     d->nbytes = 0;
 262
 263     return;
 264 }
 265
 266
 267 /* send statistics update to server (if it's time)
 268  */
 269 void
 270 tpstat_update( struct server_ctx* ctx,
 271                struct tps_data* d, ssize_t nbytes )
 272 {
 273     static const double MAX_NOUPDATE_ITER = 1000.0;
 274     static const double MAX_SEC = 10.0;
 275
 276     struct tput_stat ts;
 277     int writefd = -1, rc = 0;
 278     const int DO_NOT_SET_PID = 0;
 279     ssize_t nwr = -1;
 280     double nsec;
 281
 282     assert( ctx && d );
 283
 284     d->nbytes += nbytes;
 285
 286     nsec = difftime( time(NULL), d->tm_from );
 287     d->niter++;
 288     if( !((d->niter >= MAX_NOUPDATE_ITER) || (nsec >= MAX_SEC)) )
 289         return;
 290
 291     /* attempt to send update to server */
 292     d->niter = 0;
 293
 294     ts.sender_id = d->pid;
 295     ts.nbytes = d->nbytes;
 296     ts.nsec = nsec;
 297
 298     writefd = ctx->cpipe[1];
 299     do {
 300         nwr = write( writefd, &ts, sizeof(ts) );
 301         if( nwr <= 0 ) {
 302             if( (EINTR != errno) && (EAGAIN != errno) ) {
 303                 mperror( g_flog, errno, "%s: write", __func__ );
 304                 rc = -1;
 305                 break;
 306             }
 307             /* if it's an interrupt or pipe full - ignore */
 308             TRACE( (void)tmfprintf( g_flog, "%s - write error [%s] - ignored\n",
 309                             __func__, strerror(errno)) );
 310             break;
 311         }
 312         if( sizeof(ts) != (size_t)nwr ) {
 313             (void)tmfprintf( g_flog, "%s - wrote [%d] bytes to pipe, "
 314                     "expected [%u]\n",
 315                     __func__, nwr, (int)sizeof(ts) );
 316             rc = -1;
 317             break;
 318         }
 319
 320         /*
 321         TRACE( (void)tmfprintf( g_flog,
 322                 "Sent TSTAT={ sender=[%ld], bytes=[%f], seconds=[%f] }\n",
 323                 (long)ts.sender_id, ts.nbytes, ts.nsec) );
 324         */
 325     } while(0);
 326
 327
 328     if( 0 == rc ) {
 329         tpstat_init( d, DO_NOT_SET_PID );
 330     }
 331
 332     return;
 333 }
 334
 335
 336 /* read client statistics data and update the context
 337  */
 338 int
 339 tpstat_read( struct server_ctx* ctx )
 340 {
 341     int cindex = -1;
 342     int readfd = -1;
 343     ssize_t nread = 0;
 344     struct tput_stat ts;
 345     struct client_ctx* client = NULL;
 346
 347     assert( ctx );
 348
 349     readfd = ctx->cpipe[0];
 350     assert( readfd > 0 );
 351
 352     (void)memset( &ts, 0, sizeof(ts) );
 353
 354     nread = read( readfd, &ts, sizeof(ts) );
 355     if( nread <= 0 ) {
 356         if( (EINTR != errno) && (EAGAIN != errno) ) {
 357             mperror( g_flog, errno, "%s: read", __func__ );
 358             return ERR_INTERNAL;
 359         }
 360         /* if it's an interrupt or no data available - ignore */
 361         TRACE( (void)tmfprintf( g_flog, "%s - read error [%s] - ingored\n",
 362                         __func__, strerror(errno)) );
 363         return 0;
 364     }
 365     if( sizeof(ts) != (size_t)nread ) {
 366         (void)tmfprintf( g_flog, "%s - read [%d] bytes from pipe, expected [%u]\n",
 367                 __func__, nread, (int)sizeof(ts) );
 368         return ERR_INTERNAL;
 369     }
 370
 371     TRACE( (void)tmfprintf( g_flog,
 372             "Received TSTAT={ sender=[%ld], bytes=[%f], seconds=[%f] }\n",
 373             (long)ts.sender_id, ts.nbytes, ts.nsec) );
 374
 375     cindex = find_client( ctx, (pid_t)ts.sender_id );
 376     if( -1 == cindex ) {
 377         (void)tmfprintf( g_flog, "%s - cannot find client [%ld]\n",
 378                 __func__, (long)ts.sender_id );
 379         /* ignore invalid client id's */
 380         return 0;
 381     }
 382
 383     client = &(ctx->cl[ cindex ]);
 384     client->tstat = ts;
 385
 386     TRACE( (void)tmfprintf( g_flog, "Updated context for pid=[%d]; "
 387                 "[%.1f] Kb/sec\n", client->pid, (ts.nbytes / 1024) / ts.nsec  ) );
 388     return 0;
 389 }
 390
 391
 392
 393 /* __EOF__ */
 394