// Parallelized, lossless reachability set storage -*- c++ -*-

#ifdef __GNUC__
# pragma implementation "ParSet.h"
#endif // __GNUC__

#if defined WIN32 || defined __WIN32
# undef __STRICT_ANSI__
# include "ParSet.h"
# include <sys/types.h>
# include <winsock.h>
typedef long ssize_t;
# undef HAVE_POLL
#else
# include "ParSet.h"
# include <sys/types.h>
# include <sys/socket.h>
# if defined __CYGWIN__ || defined __APPLE__
#  undef HAVE_POLL
#  include <sys/select.h>
# elif defined __OpenBSD__
#  define HAVE_POLL
extern "C" {
#  include <sys/poll.h>
};
# else
#  define HAVE_POLL
#  include <sys/poll.h>
# endif
#endif
#ifndef USE_MMAP
# include <unistd.h>
#endif // !USE_MMAP

#include <stdlib.h> // exit(2)
#ifndef __WIN32
# include <sys/errno.h>
#endif // !__WIN32
#include <errno.h>

#ifndef MSG_NOSIGNAL
# define MSG_NOSIGNAL 0
#endif // MSG_NOSIGNAL

/** @file ParSet.C
 * Parallelized, lossless reachability set storage
 */

/* Copyright  2002-2003 Marko Mkel (msmakela@tcs.hut.fi).

   This file is part of MARIA, a reachability analyzer and model checker
   for high-level Petri nets.

   MARIA is free software; you can redistribute it and/or modify it
   under the terms of the GNU General Public License as published by
   the Free Software Foundation; either version 2, or (at your option)
   any later version.

   MARIA is distributed in the hope that it will be useful, but
   WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   General Public License for more details.

   The GNU General Public License is often shipped with GNU software, and
   is generally kept in a file called COPYING or LICENSE.  If you do not
   have a copy of the license, write to the Free Software Foundation,
   59 Temple Place, Suite 330, Boston, MA 02111 USA. */

ParSet::ParSet (int s) : StateSet (),
  myFD (s), myRecv (), mySend (), myRecvCount (0), mySendCount (0)
{
}

ParSet::~ParSet ()
{
  close (myFD);
}

/** Send pending data to a socket
 * @param fd	the socket to be written
 * @param sbuf	the buffer to write
 * @param sbufu	index of the first unsent byte in sbuf
 */
static void
sendBuf (int fd,
	 class BytePacker& sbuf,
	 unsigned& sbufu)
{
  assert (sbuf.getLength () > sbufu);
  const char* buf = reinterpret_cast<const char*>(sbuf.getBuf ());
  ssize_t i = send (fd, buf + sbufu, sbuf.getLength () - sbufu,
		    MSG_NOSIGNAL);
  if (i < 0) {
#if defined WIN32 || defined __WIN32
    fprintf (stderr, "send: WinSock error %d\n", WSAGetLastError ());
#else
    perror ("send");
#endif
    exit (-1);
  }
  if ((sbufu += i) == sbuf.getLength ())
    sbuf.clear (), sbufu = 0;
}

void
ParSet::reject (const void* dstate,
		size_t dlen,
		const class StateSetReporter&,
		enum StateSet::Code reason,
		bool reduced) const
{
  assert (!dstate == !dlen);
  mySend.append (reduced ? unsigned (reason) | 0x20 : unsigned (reason));
  mySend.append (dlen);
  mySend.append (dstate, dlen);
  // flush the buffer to notify the server of the rejected state
  do
    sendBuf (myFD, mySend, mySendCount);
  while (mySendCount);
}

bool
ParSet::do_add (const void* buf,
		size_t size)
{
  mySearch.push (buf, size, 0);
  return true;
}

word_t*
ParSet::pop (bool, size_t& size)
{
  if (unsigned numStates = mySearch.size ()) {
    if (getNumArcs ())
      mySend.append (addStates), mySend.append (numStates);
    else
      assert (numStates == 1), mySend.append (initialState);
    for (StateList::const_iterator i = mySearch.begin ();
	 i != mySearch.end (); i++)
      mySend.append (i->size), mySend.append (i->data, i->size);
  }
  else
    mySend.append (addStates), mySend.append (0);

  mySearch.clear ();
 again:
#ifndef HAVE_POLL
  fd_set readfds, writefds;
  FD_ZERO (&readfds); FD_ZERO (&writefds);
  FD_SET (myFD, &readfds); FD_SET (myFD, &writefds);
  switch (select (myFD + 1, &readfds, &writefds, 0, 0)) {
  default:
#if defined WIN32 || defined __WIN32
    if (WSAGetLastError () == WSAEINTR)
      goto again;
    fprintf (stderr, "select: WinSock error %d\n", WSAGetLastError ());
#else
    if (errno == EINTR)
      goto again;
    perror ("select");
#endif
    return 0;
  case 1:
  case 2:
    if (FD_ISSET (myFD, &writefds))
      sendBuf (myFD, mySend, mySendCount);
    if (FD_ISSET (myFD, &readfds))
      goto enqueue;
    // fall through: no data available for reading
  case 0:
    goto dequeue;
  }
#else
  struct pollfd ufd;
  ufd.fd = myFD;
  ufd.events = POLLIN | POLLOUT;
  ufd.revents = 0;
  switch (poll (&ufd, 1, -1)) {
  default:
    if (errno == EINTR)
      goto again;
    perror ("poll");
    return 0;
  case 1:
    if (ufd.revents & POLLOUT)
      sendBuf (myFD, mySend, mySendCount);
    if (ufd.revents & POLLIN)
      goto enqueue;
    // fall through: no data available for reading
  case 0:
    goto dequeue;
  }
#endif

 dequeue:
  {
    class ByteUnpacker u (myRecv.getBuf () + myRecvCount);
    unsigned len;
    if (!u.extract (myRecv, len) || !u.ensureData (myRecv, len)) {
      assert (myRecv.getBuf () + myRecvCount <= u.buf);
      goto enqueue;
    }
    if ((myRecvCount = u.buf - myRecv.getBuf () + len) == myRecv.getLength ())
      myRecvCount = 0, myRecv.clear ();

    size = len;
    return static_cast<word_t*>
      (memcpy (new word_t[1 + (len - 1) / sizeof (word_t)], u.buf, len));
  }

 enqueue:
  assert (myRecvCount
	  ? myRecvCount < myRecv.getLength ()
	  : !myRecv.getLength ());
  myRecv.allocate (2048);

  ssize_t rlen = recv (myFD, reinterpret_cast<char*>
		       (myRecv.getBuf () + myRecv.getLength ()),
		       myRecv.getAllocated () - myRecv.getLength (),
		       MSG_NOSIGNAL);
  if (rlen < 0) {
#if defined WIN32 || defined __WIN32
    int wsaerror = WSAGetLastError ();
    if (wsaerror != WSAECONNRESET)
      fprintf (stderr, "recv: WinSock error %d\n", wsaerror);
#else
    if (errno != ECONNRESET)
      perror ("recv");
#endif
    return 0;
  }
  else if (!rlen) // the remote end tells us to terminate
    return 0;

  myRecv.setLength (myRecv.getLength () + rlen);
  goto dequeue;
}