xdasd_net.c

Go to the documentation of this file.

/*----------------------------------------------------------------------------
 * Copyright (c) 2006, Novell, Inc.
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without 
 * modification, are permitted provided that the following conditions are 
 * met:
 * 
 *     * Redistributions of source code must retain the above copyright 
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright 
 *       notice, this list of conditions and the following disclaimer in the 
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the Novell nor the names of its contributors 
 *       may be used to endorse or promote products derived from this 
 *       software without specific prior written permission.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 
 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR 
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *--------------------------------------------------------------------------*/

#include <xdas_base.h>
#include <xdasd.h>

#include "xdasd_filter.h"
#include "xdasd_event.h"
#include "xdasd_list.h"
#include "xdasd_net.h"
#include "xdasd_log.h"

#include <malloc.h>
#include <string.h>
#include <errno.h>
#include <stdio.h>

#ifdef _WIN32
# define socklen_t int
# define so_bool_t char
# define BAD_SOCKET INVALID_SOCKET

/* map WSA error codes to unix names */
# define EWOULDBLOCK        WSAEWOULDBLOCK
# define EINPROGRESS        WSAEINPROGRESS
# define EALREADY           WSAEALREADY
# define ENOTSOCK           WSAENOTSOCK
# define EDESTADDRREQ       WSAEDESTADDRREQ
# define EMSGSIZE           WSAEMSGSIZE
# define EPROTOTYPE         WSAEPROTOTYPE
# define ENOPROTOOPT        WSAENOPROTOOPT
# define EPROTONOSUPPORT    WSAEPROTONOSUPPORT
# define ESOCKTNOSUPPORT    WSAESOCKTNOSUPPORT
# define EOPNOTSUPP         WSAEOPNOTSUPP
# define EPFNOSUPPORT       WSAEPFNOSUPPORT
# define EAFNOSUPPORT       WSAEAFNOSUPPORT
# define EADDRINUSE         WSAEADDRINUSE
# define EADDRNOTAVAIL      WSAEADDRNOTAVAIL
# define ENETDOWN           WSAENETDOWN
# define ENETUNREACH        WSAENETUNREACH
# define ENETRESET          WSAENETRESET
# define ECONNABORTED       WSAECONNABORTED
# define ECONNRESET         WSAECONNRESET
# define ENOBUFS            WSAENOBUFS
# define EISCONN            WSAEISCONN
# define ENOTCONN           WSAENOTCONN
# define ESHUTDOWN          WSAESHUTDOWN
# define ETOOMANYREFS       WSAETOOMANYREFS
# define ETIMEDOUT          WSAETIMEDOUT
# define ECONNREFUSED       WSAECONNREFUSED
# define ELOOP              WSAELOOP
# define EHOSTDOWN          WSAEHOSTDOWN
# define EHOSTUNREACH       WSAEHOSTUNREACH
# define EREMOTE            WSAEREMOTE

#else
# include <sys/types.h>
# include <sys/socket.h>
# include <sys/un.h>
# include <arpa/inet.h>
# include <unistd.h>
# include <fcntl.h>
# define so_bool_t int
# define BAD_SOCKET -1
# define closesocket close
#endif

#define SOCKET_UNUSED      0
#define SOCKET_CLOSE       1
#define SOCKET_LISTEN      2
#define STREAM_READ_FIRST  3
#define STREAM_READ        4
#define STREAM_WRITE_FIRST 5
#define STREAM_WRITE       6
/* @} */

typedef struct XDASD_Socket_tag
{
   XDLItem item;
   int state;
   sockfd_t fd;
   struct sockaddr_storage localaddr;
   struct sockaddr_storage peeraddr;
   xdas_buffer recvbuf;
   xdas_buffer sendbuf;
} XDASD_Socket;

static XDList s_socklist = {0, 0, 0};

static XDASD_Socket * xdasd_socket_alloc(void)
{
   XDASD_Socket * sock;

   if ((sock = (XDASD_Socket *)malloc(sizeof(*sock))) != 0)
   {
      memset(sock, 0, sizeof(*sock));
      sock->fd = BAD_SOCKET;
   }
   return sock;
}

static void xdasd_socket_free(XDASD_Socket * sock)
{
   if (sock->fd != BAD_SOCKET)
      closesocket(sock->fd);

   xdas_buffer_free(sock->recvbuf);
   xdas_buffer_free(sock->sendbuf);

   free(sock);
}

static int xdasd_socket_error(void)
{
#ifdef _WIN32
   return (int)WSAGetLastError();
#else
   return errno;
#endif
}

static int xdasd_set_nonblocking(sockfd_t fd)
{
   int err = 0;
#ifdef _WIN32
   {
      u_long fdflags = 1;
      if ((err = ioctlsocket(fd, FIONBIO, &fdflags)) != 0)
         errno = WSAGetLastError();
   }
#else
   {
      int fdflags = fcntl(fd, F_GETFL, 0);
      fcntl(fd, F_SETFL, fdflags | O_NONBLOCK);
   }
#endif
   return err? -1: 0;
}

static XDASD_Socket * xdasd_create_listen_socket(const char * ipcpath)
{
   XDASD_Socket * sock;

   if ((sock = xdasd_socket_alloc()) != 0)
   {
      struct sockaddr_storage ss;
      int addrsz = sizeof(ss);

      memset(&ss, 0, sizeof(ss));

      if (!ipcpath)
      {
         struct sockaddr_in * v4 = (struct sockaddr_in *)&ss;
         v4->sin_family = AF_INET;
         v4->sin_port = htons(XDASD_RESERVED_PORT);
         v4->sin_addr.s_addr = htonl(INADDR_LOOPBACK);
         addrsz = sizeof(*v4);
      }
#ifndef _WIN32
      else
      {
         struct sockaddr_un * ud = (struct sockaddr_un *)&ss;
         ud->sun_family = AF_LOCAL;
         strncpy(ud->sun_path, ipcpath, sizeof(ud->sun_path) - 1);
         addrsz = sizeof(*ud);
      }
#endif

      if ((sock->fd = socket(ss.ss_family, SOCK_STREAM, 0)) != BAD_SOCKET)
      {
         so_bool_t reuse = 1;

         setsockopt(sock->fd, SOL_SOCKET, SO_REUSEADDR, 
               (char *)&reuse, sizeof(reuse));

         if (bind(sock->fd, (struct sockaddr *)&ss, addrsz) == 0)
         {
            if (listen(sock->fd, 5) == 0)
            {
               memcpy(&sock->localaddr, &ss, addrsz);
               xdasd_set_nonblocking(sock->fd);
               sock->state = SOCKET_LISTEN;
               return sock;
            }
         }
      }
   }

   /* There's a problem if we get to here - clean up and return null. */
   if (sock)
      xdasd_socket_free(sock);
   return 0;
}

static void xdasd_accept(XDASD_Socket * sock)
{
   sockfd_t fd;
   XDASD_Socket * connsock;
   struct sockaddr_storage peeraddr;
   socklen_t peeraddrlen = sizeof(peeraddr);

   if ((fd = accept(sock->fd, (struct sockaddr *)&peeraddr, 
         &peeraddrlen)) != BAD_SOCKET)
   {
      xdasd_set_nonblocking(sock->fd);
      if ((connsock = xdasd_socket_alloc()) != 0)
      {
         connsock->fd = fd;
         memcpy(&connsock->peeraddr, &peeraddr, peeraddrlen);
         memcpy(&connsock->localaddr, &peeraddr, peeraddrlen);
         connsock->state = STREAM_READ_FIRST;
         xdasd_list_link_head(&s_socklist, &connsock->item);
      }
   }
}

static int xdasd_net_dispatch(XDASD_Socket * sock)
{
   static int (*handler_table[])(int*,xdas_buffer,xdas_buffer*) = 
   {
      xdasd_event_process,
      xdasd_filter_process,
   };

   size_t reqsz, rspsz;
   unsigned reqfn;
   int rspst, rspms = 0;
   xdas_buffer rsp, req = sock->recvbuf;

   if ((rsp = xdas_buffer_realloc(sock->sendbuf, 12)) == 0)
   {
      return (errno = ENOMEM), -1;
   }

   /* reserve space for 3 response header fields */
   rsp->curpos = rsp->start + 12;

   /* ensure request buffer contains a full request header */
   reqsz = xdas_buffer_get_uint32(req);
   if (req->end - req->curpos < 4 || (reqfn = xdas_buffer_get_uint32(req)) 
         >= xdas_elemcount(handler_table))
   {
      xdas_buffer_put_uint32(rsp, (unsigned)(rsp->curpos - rsp->start));
      xdas_buffer_put_uint32(rsp, XDAS_S_FAILURE);
      xdas_buffer_put_uint32(rsp, OXDAS_MS_PROTOCOL);
      sock->sendbuf = rsp;
      return 0;
   }

   /* printf("REQ: %d - %u bytes.\n", reqfn, reqsz); */

   /* dispatch to appropriate request handler */
   rspst = handler_table[reqfn](&rspms, req, &rsp);

   /* fill in response buffer size and response code */
   rspsz = rsp->curpos - rsp->start;
   rsp->curpos = rsp->start;
   xdas_buffer_put_uint32(rsp, rspsz);
   xdas_buffer_put_uint32(rsp, rspst);
   xdas_buffer_put_uint32(rsp, rspms);
   sock->sendbuf = rsp;

   /* printf("RSP: %d:%d - %u bytes.\n", rspst, rspms, rsp->curpos - rsp->start); */

   return 0;
}

static void xdasd_write_stream(XDASD_Socket * sock)
{
   if (sock->state == STREAM_WRITE_FIRST)
   {
      sock->sendbuf->curpos = sock->sendbuf->start;
      sock->state = STREAM_WRITE;
   }

   /* either finish the STREAM_READ_FIRST, or do the next STREAM_READ */
   if (sock->sendbuf->end - sock->sendbuf->start != 0)
   {
      int byteswritten, flags = 0;
#ifdef MSG_DONTWAIT
      flags = MSG_DONTWAIT;
#endif
      if ((byteswritten = send(sock->fd, (char *)sock->sendbuf->curpos,
            (int)(sock->sendbuf->end - sock->sendbuf->start), flags)) > 0)
      {
         sock->sendbuf->curpos += byteswritten;
         if (sock->sendbuf->curpos == sock->sendbuf->end)
            sock->state = STREAM_READ_FIRST;
      }
      else if (xdasd_socket_error() != EWOULDBLOCK)
         sock->state = SOCKET_CLOSE;
   }
}

static void xdasd_read_stream(XDASD_Socket * sock)
{
   int bytesread;

   if (sock->state == STREAM_READ_FIRST)
   {
      xdas_buffer req;
      size_t msgsz;
      char peek[4];

      if ((bytesread = recv(sock->fd, peek, sizeof(peek), MSG_PEEK)) <= 0)
      {
         sock->state = SOCKET_CLOSE;   /* connection reset by peer */
         return;
      }

      msgsz = AS_UINT32(peek);
      if (msgsz > MAX_XDAS_MSG_SIZE)
      {
         xdasd_log(0, "PROTOCOL_ERROR - message too large!\n");
         sock->state = SOCKET_CLOSE;
         return;
      }
      if ((req = xdas_buffer_realloc(sock->recvbuf, msgsz)) == 0)
      {
         xdasd_log(0, "INTERNAL_ERROR - out of memory!\n");
         sock->state = SOCKET_CLOSE;   /* out of memory */
         return;
      }
      req->curpos = req->start;
      sock->recvbuf = req;
      sock->state = STREAM_READ;
   }

   /* either finish the STREAM_READ_FIRST, or do the next STREAM_READ */
   if ((bytesread = recv(sock->fd, (char *)sock->recvbuf->curpos,
         (int)(sock->recvbuf->end - sock->recvbuf->curpos), 0)) <= 0)
   {
      sock->state = SOCKET_CLOSE;      /* connection reset by peer */
      return;
   }

   sock->recvbuf->curpos += bytesread;
   if (sock->recvbuf->curpos == sock->recvbuf->end)
   {
      /* reset curpos for data extraction, end ptr indicates reqsz. */
      sock->recvbuf->curpos = sock->recvbuf->start;
      if (xdasd_net_dispatch(sock) < 0)
      {
         sock->state = SOCKET_CLOSE;   /* ill-formatted request */
         return;
      }

      sock->state = STREAM_WRITE_FIRST;
      xdasd_write_stream(sock);
   }
}

/*--------------------------------------------------------------------------*/

int xdasd_net_init(const char * ipcpath)
{
   XDASD_Socket * sock;

#ifdef _WIN32
   {
      WORD vreq = MAKEWORD(2,2);
      WSADATA wsadata;
      if (WSAStartup(vreq, &wsadata) != 0)
         return -1;
   }
#endif

   /* remove all inbound sockets from the list */
   while (s_socklist.count)
      xdasd_socket_free((XDASD_Socket *)
            xdasd_list_unlink(&s_socklist, s_socklist.head));

#ifndef _WIN32
   /* open unix domain socket first */
   if ((sock = xdasd_create_listen_socket(ipcpath)) != 0)
   {
      xdasd_list_link_tail(&s_socklist, &sock->item);
      xdasd_log(0, "Listening on Unix domain socket...\n");
   }
   else
      xdasd_log(0, "NETWORK: Could not initialize Unix domain socket.\n");
#else
   (void)ipcpath;
#endif

   /* now open local socket */
   if ((sock = xdasd_create_listen_socket(0)) != 0)
   {
      xdasd_list_link_tail(&s_socklist, &sock->item);
      xdasd_log(0, "Listening on TCP socket...\n");
   }
   else
      xdasd_log(0, "NETWORK: Could not initialize TCP socket.\n");

#ifdef _WIN32
   if (!s_socklist.count)
   {
      errno = WSAGetLastError();
      WSACleanup();
   }
#endif

   return s_socklist.count? 0: -1;
}

#ifdef _WIN32
# pragma warning(push)
# pragma warning(disable: 4127) /* FD_SET - conditional expr is constant */
#endif

void xdasd_net_load_fdsets(int * highfd, fd_set * readfds, fd_set * writefds)
{
   XDASD_Socket * del = 0;
   XDASD_Socket * sock = (XDASD_Socket *)s_socklist.head;

   while (sock)
   {
      if (sock->fd > (sockfd_t)*highfd)
         *highfd = (int)sock->fd;

      switch(sock->state)
      {
         case SOCKET_LISTEN:
            FD_SET(sock->fd, readfds);
            break;

         case STREAM_READ_FIRST:
         case STREAM_READ:
            FD_SET(sock->fd, readfds);
            break;

         case STREAM_WRITE_FIRST:
         case STREAM_WRITE:
            FD_SET(sock->fd, writefds);
            break;

         case SOCKET_CLOSE:
            del = sock;
            break;
      }
      sock = (XDASD_Socket *)sock->item.next;
      if (del)
      {
         xdasd_socket_free((XDASD_Socket *)
               xdasd_list_unlink(&s_socklist, (XDLItem *)del));
         del = 0;
      }
   }
}

#ifdef _WIN32
# pragma warning(pop)
#endif

void xdasd_net_event_handler(int fdcount, fd_set * readfds, fd_set * writefds)
{
   XDASD_Socket * sock = (XDASD_Socket *)s_socklist.head;

   while (sock && fdcount)
   {
      if (FD_ISSET(sock->fd, readfds))
      {
         switch (sock->state)
         {
            case SOCKET_LISTEN:
               xdasd_accept(sock);
               break;

            case STREAM_READ_FIRST:
            case STREAM_READ:
               xdasd_read_stream(sock);
               break;

         }
         --fdcount;
      }
      else if (FD_ISSET(sock->fd, writefds))
      {
         switch (sock->state)
         {
            case STREAM_WRITE_FIRST:
            case STREAM_WRITE:
               xdasd_write_stream(sock);
               break;
         }
         --fdcount;
      }
      sock = (XDASD_Socket *)sock->item.next;
   }
}

void xdasd_net_exit(void)
{
   XDASD_Socket * sock = (XDASD_Socket *)s_socklist.head;
   while (sock)
   {
      XDASD_Socket * del = sock;
      sock = (XDASD_Socket *)sock->item.next;
      xdasd_socket_free(del);
   } 
   s_socklist.head = s_socklist.tail = 0;
   s_socklist.count = 0;

#ifdef _WIN32
   WSACleanup();
#endif
}