IO-AIO/schmorp.h

#ifndef SCHMORP_PERL_H_
#define SCHMORP_PERL_H_

/* WARNING
 * This header file is a shared resource between many modules.
 */

#include <signal.h>

#ifndef _WIN32
# include <poll.h>
#endif

/* useful stuff, used by schmorp mostly */

#include "patchlevel.h"

#define PERL_VERSION_ATLEAST(a,b,c)                             \
  (PERL_REVISION > (a)                                          \
   || (PERL_REVISION == (a)                                     \
       && (PERL_VERSION > (b)                                   \
           || (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))

#if !PERL_VERSION_ATLEAST (5,6,0)
# ifndef PL_ppaddr
#  define PL_ppaddr ppaddr
# endif
# ifndef call_sv
#  define call_sv perl_call_sv
# endif
# ifndef get_sv
#  define get_sv perl_get_sv
# endif
# ifndef get_cv
#  define get_cv perl_get_cv
# endif
# ifndef IS_PADGV
#  define IS_PADGV(v) 0
# endif
# ifndef IS_PADCONST
#  define IS_PADCONST(v) 0
# endif
#endif

/* 5.11 */
#ifndef CxHASARGS
# define CxHASARGS(cx) (cx)->blk_sub.hasargs
#endif

/* 5.10.0 */
#ifndef SvREFCNT_inc_NN
# define SvREFCNT_inc_NN(sv) SvREFCNT_inc (sv)
#endif

/* 5.8.8 */
#ifndef GV_NOTQUAL
# define GV_NOTQUAL 0
#endif
#ifndef newSV
# define newSV(l) NEWSV(0,l)
#endif
#ifndef CvISXSUB_on
# define CvISXSUB_on(cv) (void)cv
#endif
#ifndef CvISXSUB
# define CvISXSUB(cv) (CvXSUB (cv) ? TRUE : FALSE)
#endif
#ifndef Newx
# define Newx(ptr,nitems,type) New (0,ptr,nitems,type)
#endif

/* 5.8.7 */
#ifndef SvRV_set
# define SvRV_set(s,v) SvRV(s) = (v)
#endif

static int
s_signum (SV *sig)
{
#ifndef SIG_SIZE
  /* kudos to Slaven Rezic for the idea */
  static char sig_size [] = { SIG_NUM };
# define SIG_SIZE (sizeof (sig_size) + 1)
#endif
  dTHX;
  int signum;

  SvGETMAGIC (sig);

  for (signum = 1; signum < SIG_SIZE; ++signum)
    if (strEQ (SvPV_nolen (sig), PL_sig_name [signum]))
      return signum;

  signum = SvIV (sig);

  if (signum > 0 && signum < SIG_SIZE)
    return signum;

  return -1;
}

static int
s_signum_croak (SV *sig)
{
  int signum = s_signum (sig);

  if (signum < 0)
    {
      dTHX;
      croak ("%s: invalid signal name or number", SvPV_nolen (sig));
    }

  return signum;
}

static int
s_fileno (SV *fh, int wr)
{
  dTHX;
  SvGETMAGIC (fh);

  if (SvROK (fh))
    {
      fh = SvRV (fh);
      SvGETMAGIC (fh);
    }

  if (SvTYPE (fh) == SVt_PVGV)
    return PerlIO_fileno (wr ? IoOFP (sv_2io (fh)) : IoIFP (sv_2io (fh)));

  if (SvOK (fh) && (SvIV (fh) >= 0) && (SvIV (fh) < 0x7fffffffL))
    return SvIV (fh);

  return -1;
}

static int
s_fileno_croak (SV *fh, int wr)
{
  int fd = s_fileno (fh, wr);

  if (fd < 0)
    {
      dTHX;
      croak ("%s: illegal fh argument, either not an OS file or read/write mode mismatch", SvPV_nolen (fh));
    }

  return fd;
}

static SV *
s_get_cv (SV *cb_sv)
{
  dTHX;
  HV *st;
  GV *gvp;

  return (SV *)sv_2cv (cb_sv, &st, &gvp, 0);
}

static SV *
s_get_cv_croak (SV *cb_sv)
{
  SV *cv = s_get_cv (cb_sv);

  if (!cv)
    {
      dTHX;
      croak ("%s: callback must be a CODE reference or another callable object", SvPV_nolen (cb_sv));
    }

  return cv;
}

/*****************************************************************************/
/* gensub: simple closure generation utility */

#define S_GENSUB_ARG CvXSUBANY (cv).any_ptr

/* create a closure from XS, returns a code reference */
/* the arg can be accessed via GENSUB_ARG from the callback */
/* the callback must use dXSARGS/XSRETURN */
static SV *
s_gensub (pTHX_ void (*xsub)(pTHX_ CV *), void *arg)
{
  CV *cv = (CV *)newSV (0);

  sv_upgrade ((SV *)cv, SVt_PVCV);

  CvANON_on (cv);
  CvISXSUB_on (cv);
  CvXSUB (cv) = xsub;
  S_GENSUB_ARG = arg;

  return newRV_noinc ((SV *)cv);
}

/** portable pipe/socketpair */

#ifdef USE_SOCKETS_AS_HANDLES
# define S_TO_SOCKET(x) (win32_get_osfhandle (x))
#else
# define S_TO_SOCKET(x) (x)
#endif

#ifdef _WIN32
/* taken almost verbatim from libev's ev_win32.c */
/* oh, the humanity! */
static int
s_pipe (int filedes [2])
{
  struct sockaddr_in addr = { 0 };
  int addr_size = sizeof (addr);
  struct sockaddr_in adr2;
  int adr2_size = sizeof (adr2);
  SOCKET listener;
  SOCKET sock [2] = { -1, -1 };

  if ((listener = socket (AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET) 
    return -1;

  addr.sin_family = AF_INET;
  addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
  addr.sin_port = 0;

  if (bind (listener, (struct sockaddr *)&addr, addr_size))
    goto fail;

  if (getsockname (listener, (struct sockaddr *)&addr, &addr_size))
    goto fail;

  if (listen (listener, 1))
    goto fail;

  if ((sock [0] = socket (AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET) 
    goto fail;

  if (connect (sock [0], (struct sockaddr *)&addr, addr_size))
    goto fail;

  if ((sock [1] = accept (listener, 0, 0)) < 0)
    goto fail;

  /* windows vista returns fantasy port numbers for getpeername.
   * example for two interconnected tcp sockets:
   *
   * (Socket::unpack_sockaddr_in getsockname $sock0)[0] == 53364
   * (Socket::unpack_sockaddr_in getpeername $sock0)[0] == 53363
   * (Socket::unpack_sockaddr_in getsockname $sock1)[0] == 53363
   * (Socket::unpack_sockaddr_in getpeername $sock1)[0] == 53365
   *
   * wow! tridirectional sockets!
   *
   * this way of checking ports seems to work:
   */
  if (getpeername (sock [0], (struct sockaddr *)&addr, &addr_size))
    goto fail;

  if (getsockname (sock [1], (struct sockaddr *)&adr2, &adr2_size))
    goto fail;

  errno = WSAEINVAL;
  if (addr_size != adr2_size
      || addr.sin_addr.s_addr != adr2.sin_addr.s_addr /* just to be sure, I mean, it's windows */
      || addr.sin_port        != adr2.sin_port)
    goto fail;

  closesocket (listener);

#ifdef USE_SOCKETS_AS_HANDLES
  /* when select isn't winsocket, we also expect socket, connect, accept etc.
   * to work on fds */
  filedes [0] = sock [0];
  filedes [1] = sock [1];
#else
  filedes [0] = _open_osfhandle (sock [0], 0);
  filedes [1] = _open_osfhandle (sock [1], 0);
#endif

  return 0;

fail:
  closesocket (listener);

  if (sock [0] != INVALID_SOCKET) closesocket (sock [0]);
  if (sock [1] != INVALID_SOCKET) closesocket (sock [1]);

  return -1;
}

#define s_socketpair(domain,type,protocol,filedes) s_pipe (filedes)

static int
s_fd_blocking (int fd, int blocking)
{
  blocking = !blocking;

  return ioctlsocket (S_TO_SOCKET (fd), FIONBIO, &blocking);
}

#define s_fd_prepare(fd) s_fd_blocking (fd, 0)

#else

#define s_socketpair(domain,type,protocol,filedes) socketpair (domain, type, protocol, filedes)
#define s_pipe(filedes) pipe (filedes)

static int
s_fd_blocking (int fd, int blocking)
{
  return fcntl (fd, F_SETFL, blocking ? 0 : O_NONBLOCK);
}

static int
s_fd_prepare (int fd)
{
  return s_fd_blocking (fd, 0)
         || fcntl (fd, F_SETFD, FD_CLOEXEC);
}

#endif

#if __linux && (__GLIBC__ > 2 || (__GLIBC__ == 2 && __GLIBC_MINOR__ >= 7))
/* our minimum requirement is glibc 2.7 which has the stub, but not the header */
# include <stdint.h>
# ifdef __cplusplus
extern "C" {
# endif
  int eventfd (unsigned int initval, int flags);
# ifdef __cplusplus
}
# endif
#else
# define eventfd(initval,flags) -1
#endif

typedef struct {
  int fd[2]; /* read, write fd, might be equal */
  int len; /* write length (1 pipe/socket, 8 eventfd) */
  volatile sig_atomic_t sent;
} s_epipe;

static int
s_epipe_new (s_epipe *epp)
{
  s_epipe ep;

  ep.fd [0] = ep.fd [1] = eventfd (0, 0);

  if (ep.fd [0] >= 0)
    {
      s_fd_prepare (ep.fd [0]);
      ep.len = 8;
    }
  else
    {
      if (s_pipe (ep.fd))
        return -1;

      if (s_fd_prepare (ep.fd [0])
          || s_fd_prepare (ep.fd [1]))
        {
           close (ep.fd [0]);
           close (ep.fd [1]);
           return -1;
        }

      ep.len = 1;
    }

  ep.sent = 0;
  *epp = ep;
  return 0;
}

static void
s_epipe_destroy (s_epipe *epp)
{
  close (epp->fd [0]);

  if (epp->fd [1] != epp->fd [0])
    close (epp->fd [1]);

  epp->len = 0;
}

static void
s_epipe_signal (s_epipe *epp)
{
  if (epp->sent)
    return;

  epp->sent = 1;
#ifdef _WIN32
  send (epp->fd [1], epp, 1);
#else
  static uint64_t counter = 1;
  write (epp->fd [1], &counter, epp->len);
#endif
}

static void
s_epipe_drain (s_epipe *epp)
{
  char buf [9];

#ifdef _WIN32
  PerlSock_recv (epp->fd [0], buf, sizeof (buf), 0);
#else
  read (epp->fd [0], buf, sizeof (buf));
#endif

  epp->sent = 0;
}

/* like new, but dups over old */
static int
s_epipe_renew (s_epipe *epp)
{
  s_epipe epn;

  if (epp->fd [1] != epp->fd [0])
    close (epp->fd [1]);

  if (s_epipe_new (&epn))
    return -1;

  if (epp->len)
    {
      if (dup2 (S_TO_SOCKET (epn.fd [0]), S_TO_SOCKET (epp->fd [0])) < 0)
        croak ("unable to dup over old event pipe"); /* should not croak */

      if (epp->fd [1] != epp->fd [0])
        close (epn.fd [0]);

      epn.fd [0] = epp->fd [0];
    }

  *epp = epn;

  return 0;
}

#define s_epipe_fd(epp) ((epp)->fd [0])

static int
s_epipe_wait (s_epipe *epp)
{
#ifdef _WIN32
  fd_set rfd;

  FD_ZERO (&rfd);
  FD_SET (s_epipe_fd (epp), &rfd);

  return PerlSock_select (s_epipe_fd (epp) + 1, &rfd, 0, 0, 0);
#else
  /* poll is preferable on posix systems */
  struct pollfd pfd;

  pfd.fd = s_epipe_fd (epp);
  pfd.events = POLLIN;

  return poll (&pfd, 1, 0);
#endif
}

#endif

Revision:	1.5
Committed:	Wed Jul 15 01:36:04 2009 UTC (14 years, 10 months ago) by root
Content type:	text/plain
Branch:	MAIN
Changes since 1.4:	+290 -3 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#ifndef SCHMORP_PERL_H_
2			#define SCHMORP_PERL_H_
3
4	root	1.2	/* WARNING
5			* This header file is a shared resource between many modules.
6			*/
7
8	root	1.5	#include <signal.h>
9
10			#ifndef _WIN32
11			# include <poll.h>
12			#endif
13
14	root	1.1	/* useful stuff, used by schmorp mostly */
15
16	root	1.4	#include "patchlevel.h"
17
18	root	1.1	#define PERL_VERSION_ATLEAST(a,b,c) \
19			(PERL_REVISION > (a) \
20			\|\| (PERL_REVISION == (a) \
21			&& (PERL_VERSION > (b) \
22			\|\| (PERL_VERSION == (b) && PERL_SUBVERSION >= (c)))))
23
24			#if !PERL_VERSION_ATLEAST (5,6,0)
25			# ifndef PL_ppaddr
26			# define PL_ppaddr ppaddr
27			# endif
28			# ifndef call_sv
29			# define call_sv perl_call_sv
30			# endif
31			# ifndef get_sv
32			# define get_sv perl_get_sv
33			# endif
34			# ifndef get_cv
35			# define get_cv perl_get_cv
36			# endif
37			# ifndef IS_PADGV
38			# define IS_PADGV(v) 0
39			# endif
40			# ifndef IS_PADCONST
41			# define IS_PADCONST(v) 0
42			# endif
43			#endif
44
45			/* 5.11 */
46			#ifndef CxHASARGS
47			# define CxHASARGS(cx) (cx)->blk_sub.hasargs
48			#endif
49
50			/* 5.10.0 */
51			#ifndef SvREFCNT_inc_NN
52			# define SvREFCNT_inc_NN(sv) SvREFCNT_inc (sv)
53			#endif
54
55			/* 5.8.8 */
56			#ifndef GV_NOTQUAL
57			# define GV_NOTQUAL 0
58			#endif
59			#ifndef newSV
60			# define newSV(l) NEWSV(0,l)
61			#endif
62			#ifndef CvISXSUB_on
63			# define CvISXSUB_on(cv) (void)cv
64			#endif
65			#ifndef CvISXSUB
66			# define CvISXSUB(cv) (CvXSUB (cv) ? TRUE : FALSE)
67			#endif
68			#ifndef Newx
69			# define Newx(ptr,nitems,type) New (0,ptr,nitems,type)
70			#endif
71
72			/* 5.8.7 */
73			#ifndef SvRV_set
74			# define SvRV_set(s,v) SvRV(s) = (v)
75			#endif
76
77			static int
78			s_signum (SV *sig)
79			{
80			#ifndef SIG_SIZE
81			/* kudos to Slaven Rezic for the idea */
82			static char sig_size [] = { SIG_NUM };
83			# define SIG_SIZE (sizeof (sig_size) + 1)
84			#endif
85	root	1.5	dTHX;
86	root	1.1	int signum;
87
88			SvGETMAGIC (sig);
89
90			for (signum = 1; signum < SIG_SIZE; ++signum)
91			if (strEQ (SvPV_nolen (sig), PL_sig_name [signum]))
92			return signum;
93
94			signum = SvIV (sig);
95
96			if (signum > 0 && signum < SIG_SIZE)
97			return signum;
98
99			return -1;
100			}
101
102			static int
103			s_signum_croak (SV *sig)
104			{
105			int signum = s_signum (sig);
106
107			if (signum < 0)
108	root	1.5	{
109			dTHX;
110			croak ("%s: invalid signal name or number", SvPV_nolen (sig));
111			}
112	root	1.1
113			return signum;
114			}
115
116			static int
117			s_fileno (SV *fh, int wr)
118			{
119	root	1.5	dTHX;
120	root	1.1	SvGETMAGIC (fh);
121
122			if (SvROK (fh))
123			{
124			fh = SvRV (fh);
125			SvGETMAGIC (fh);
126			}
127
128			if (SvTYPE (fh) == SVt_PVGV)
129			return PerlIO_fileno (wr ? IoOFP (sv_2io (fh)) : IoIFP (sv_2io (fh)));
130
131			if (SvOK (fh) && (SvIV (fh) >= 0) && (SvIV (fh) < 0x7fffffffL))
132			return SvIV (fh);
133
134			return -1;
135			}
136
137			static int
138			s_fileno_croak (SV *fh, int wr)
139			{
140			int fd = s_fileno (fh, wr);
141
142			if (fd < 0)
143	root	1.5	{
144			dTHX;
145			croak ("%s: illegal fh argument, either not an OS file or read/write mode mismatch", SvPV_nolen (fh));
146			}
147	root	1.1
148			return fd;
149			}
150
151			static SV *
152			s_get_cv (SV *cb_sv)
153			{
154	root	1.5	dTHX;
155	root	1.1	HV *st;
156			GV *gvp;
157
158	root	1.4	return (SV *)sv_2cv (cb_sv, &st, &gvp, 0);
159	root	1.1	}
160
161			static SV *
162			s_get_cv_croak (SV *cb_sv)
163			{
164	root	1.3	SV *cv = s_get_cv (cb_sv);
165	root	1.1
166	root	1.3	if (!cv)
167	root	1.5	{
168			dTHX;
169			croak ("%s: callback must be a CODE reference or another callable object", SvPV_nolen (cb_sv));
170			}
171	root	1.1
172	root	1.3	return cv;
173	root	1.1	}
174
175			/*****************************************************************************/
176			/* gensub: simple closure generation utility */
177
178			#define S_GENSUB_ARG CvXSUBANY (cv).any_ptr
179
180			/* create a closure from XS, returns a code reference */
181			/* the arg can be accessed via GENSUB_ARG from the callback */
182			/* the callback must use dXSARGS/XSRETURN */
183			static SV *
184			s_gensub (pTHX_ void (xsub)(pTHX_ CV ), void *arg)
185			{
186			CV cv = (CV )newSV (0);
187
188			sv_upgrade ((SV *)cv, SVt_PVCV);
189
190			CvANON_on (cv);
191			CvISXSUB_on (cv);
192			CvXSUB (cv) = xsub;
193			S_GENSUB_ARG = arg;
194
195			return newRV_noinc ((SV *)cv);
196			}
197
198	root	1.5	/** portable pipe/socketpair */
199
200			#ifdef USE_SOCKETS_AS_HANDLES
201			# define S_TO_SOCKET(x) (win32_get_osfhandle (x))
202			#else
203			# define S_TO_SOCKET(x) (x)
204			#endif
205
206			#ifdef _WIN32
207			/* taken almost verbatim from libev's ev_win32.c */
208			/* oh, the humanity! */
209			static int
210			s_pipe (int filedes [2])
211			{
212			struct sockaddr_in addr = { 0 };
213			int addr_size = sizeof (addr);
214			struct sockaddr_in adr2;
215			int adr2_size = sizeof (adr2);
216			SOCKET listener;
217			SOCKET sock [2] = { -1, -1 };
218
219			if ((listener = socket (AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
220			return -1;
221
222			addr.sin_family = AF_INET;
223			addr.sin_addr.s_addr = htonl (INADDR_LOOPBACK);
224			addr.sin_port = 0;
225
226			if (bind (listener, (struct sockaddr *)&addr, addr_size))
227			goto fail;
228
229			if (getsockname (listener, (struct sockaddr *)&addr, &addr_size))
230			goto fail;
231
232			if (listen (listener, 1))
233			goto fail;
234
235			if ((sock [0] = socket (AF_INET, SOCK_STREAM, 0)) == INVALID_SOCKET)
236			goto fail;
237
238			if (connect (sock [0], (struct sockaddr *)&addr, addr_size))
239			goto fail;
240
241			if ((sock [1] = accept (listener, 0, 0)) < 0)
242			goto fail;
243
244			/* windows vista returns fantasy port numbers for getpeername.
245			* example for two interconnected tcp sockets:
246			*
247			* (Socket::unpack_sockaddr_in getsockname $sock0)[0] == 53364
248			* (Socket::unpack_sockaddr_in getpeername $sock0)[0] == 53363
249			* (Socket::unpack_sockaddr_in getsockname $sock1)[0] == 53363
250			* (Socket::unpack_sockaddr_in getpeername $sock1)[0] == 53365
251			*
252			* wow! tridirectional sockets!
253			*
254			* this way of checking ports seems to work:
255			*/
256			if (getpeername (sock [0], (struct sockaddr *)&addr, &addr_size))
257			goto fail;
258
259			if (getsockname (sock [1], (struct sockaddr *)&adr2, &adr2_size))
260			goto fail;
261
262			errno = WSAEINVAL;
263			if (addr_size != adr2_size
264			\|\| addr.sin_addr.s_addr != adr2.sin_addr.s_addr /* just to be sure, I mean, it's windows */
265			\|\| addr.sin_port != adr2.sin_port)
266			goto fail;
267
268			closesocket (listener);
269
270			#ifdef USE_SOCKETS_AS_HANDLES
271			/* when select isn't winsocket, we also expect socket, connect, accept etc.
272			* to work on fds */
273			filedes [0] = sock [0];
274			filedes [1] = sock [1];
275			#else
276			filedes [0] = _open_osfhandle (sock [0], 0);
277			filedes [1] = _open_osfhandle (sock [1], 0);
278			#endif
279
280			return 0;
281
282			fail:
283			closesocket (listener);
284
285			if (sock [0] != INVALID_SOCKET) closesocket (sock [0]);
286			if (sock [1] != INVALID_SOCKET) closesocket (sock [1]);
287
288			return -1;
289			}
290
291			#define s_socketpair(domain,type,protocol,filedes) s_pipe (filedes)
292
293			static int
294			s_fd_blocking (int fd, int blocking)
295			{
296			blocking = !blocking;
297
298			return ioctlsocket (S_TO_SOCKET (fd), FIONBIO, &blocking);
299			}
300
301			#define s_fd_prepare(fd) s_fd_blocking (fd, 0)
302
303			#else
304
305			#define s_socketpair(domain,type,protocol,filedes) socketpair (domain, type, protocol, filedes)
306			#define s_pipe(filedes) pipe (filedes)
307
308			static int
309			s_fd_blocking (int fd, int blocking)
310			{
311			return fcntl (fd, F_SETFL, blocking ? 0 : O_NONBLOCK);
312			}
313
314			static int
315			s_fd_prepare (int fd)
316			{
317			return s_fd_blocking (fd, 0)
318			\|\| fcntl (fd, F_SETFD, FD_CLOEXEC);
319			}
320
321			#endif
322
323			#if __linux && (__GLIBC__ > 2 \|\| (__GLIBC__ == 2 && __GLIBC_MINOR__ >= 7))
324			/* our minimum requirement is glibc 2.7 which has the stub, but not the header */
325			# include <stdint.h>
326			# ifdef __cplusplus
327			extern "C" {
328			# endif
329			int eventfd (unsigned int initval, int flags);
330			# ifdef __cplusplus
331			}
332			# endif
333			#else
334			# define eventfd(initval,flags) -1
335			#endif
336
337			typedef struct {
338			int fd[2]; /* read, write fd, might be equal */
339			int len; /* write length (1 pipe/socket, 8 eventfd) */
340			volatile sig_atomic_t sent;
341			} s_epipe;
342
343			static int
344			s_epipe_new (s_epipe *epp)
345			{
346			s_epipe ep;
347
348			ep.fd [0] = ep.fd [1] = eventfd (0, 0);
349
350			if (ep.fd [0] >= 0)
351			{
352			s_fd_prepare (ep.fd [0]);
353			ep.len = 8;
354			}
355			else
356			{
357			if (s_pipe (ep.fd))
358			return -1;
359
360			if (s_fd_prepare (ep.fd [0])
361			\|\| s_fd_prepare (ep.fd [1]))
362			{
363			close (ep.fd [0]);
364			close (ep.fd [1]);
365			return -1;
366			}
367
368			ep.len = 1;
369			}
370
371			ep.sent = 0;
372			*epp = ep;
373			return 0;
374			}
375
376			static void
377			s_epipe_destroy (s_epipe *epp)
378			{
379			close (epp->fd [0]);
380
381			if (epp->fd [1] != epp->fd [0])
382			close (epp->fd [1]);
383
384			epp->len = 0;
385			}
386
387			static void
388			s_epipe_signal (s_epipe *epp)
389			{
390			if (epp->sent)
391			return;
392
393			epp->sent = 1;
394			#ifdef _WIN32
395			send (epp->fd [1], epp, 1);
396			#else
397			static uint64_t counter = 1;
398			write (epp->fd [1], &counter, epp->len);
399			#endif
400			}
401
402			static void
403			s_epipe_drain (s_epipe *epp)
404			{
405			char buf [9];
406
407			#ifdef _WIN32
408			PerlSock_recv (epp->fd [0], buf, sizeof (buf), 0);
409			#else
410			read (epp->fd [0], buf, sizeof (buf));
411			#endif
412
413			epp->sent = 0;
414			}
415
416			/* like new, but dups over old */
417			static int
418			s_epipe_renew (s_epipe *epp)
419			{
420			s_epipe epn;
421
422			if (epp->fd [1] != epp->fd [0])
423			close (epp->fd [1]);
424
425			if (s_epipe_new (&epn))
426			return -1;
427
428			if (epp->len)
429			{
430			if (dup2 (S_TO_SOCKET (epn.fd [0]), S_TO_SOCKET (epp->fd [0])) < 0)
431			croak ("unable to dup over old event pipe"); /* should not croak */
432
433			if (epp->fd [1] != epp->fd [0])
434			close (epn.fd [0]);
435
436			epn.fd [0] = epp->fd [0];
437			}
438
439			*epp = epn;
440
441			return 0;
442			}
443
444			#define s_epipe_fd(epp) ((epp)->fd [0])
445
446			static int
447			s_epipe_wait (s_epipe *epp)
448			{
449			#ifdef _WIN32
450			fd_set rfd;
451
452			FD_ZERO (&rfd);
453			FD_SET (s_epipe_fd (epp), &rfd);
454
455			return PerlSock_select (s_epipe_fd (epp) + 1, &rfd, 0, 0, 0);
456			#else
457			/* poll is preferable on posix systems */
458			struct pollfd pfd;
459
460			pfd.fd = s_epipe_fd (epp);
461			pfd.events = POLLIN;
462
463			return poll (&pfd, 1, 0);
464			#endif
465			}
466
467	root	1.1	#endif
468