Coro-Mysql/Mysql.xs

#include <sys/errno.h>
#include <unistd.h>
#include <fcntl.h>

#include <mysql.h>

#include "EXTERN.h"
#include "perl.h"
#include "XSUB.h"

#define IN_DESTRUCT PL_dirty

typedef U16 uint16;

/* cached function gv's */
static CV *readable, *writable;

#include "violite.h"

#define DESC_OFFSET 22

#define CoMy_MAGIC 0x436f4d79

typedef struct {
  int magic;
  SV *corohandle_sv, *corohandle;
  int bufofs, bufcnt;
  char buf[VIO_READ_BUFFER_SIZE];
} ourdata;

#define OURDATAPTR (*((ourdata **)((vio)->desc + DESC_OFFSET)))

static int
our_read (Vio *vio, xgptr p, int len)
{
  ourdata *our = OURDATAPTR;

  if (!our->bufcnt)
    {
      int rd;
      my_bool dummy;

      vio->vioblocking (vio, 0, &dummy);

      for (;;)
        {
          rd = recv (vio->sd, our->buf, sizeof (our->buf), 0);

          if (rd >= 0 || errno != EAGAIN)
            break;

          {
            dSP;
            PUSHMARK (SP);
            XPUSHs (our->corohandle);
            PUTBACK;
            call_sv ((SV *)readable, G_VOID | G_DISCARD);
          }
        }

      if (rd <= 0)
        return rd;

      our->bufcnt = rd;
      our->bufofs = 0;
    }

  if (our->bufcnt < len)
    len = our->bufcnt;

  memcpy (p, our->buf + our->bufofs, len);
  our->bufofs += len;
  our->bufcnt -= len;

  return len;
}

static int
our_write (Vio *vio, const xgptr p, int len)
{
  char *ptr = (char *)p;
  my_bool dummy;

  vio->vioblocking (vio, 0, &dummy);

  while (len > 0)
    {
      int wr = send (vio->sd, ptr, len, 0);

      if (wr > 0)
        {
          ptr += wr;
          len -= wr;
        }
      else if (errno == EAGAIN)
        {
          dSP;
          PUSHMARK (SP);
          XPUSHs (OURDATAPTR->corohandle);
          PUTBACK;
          call_sv ((SV *)writable, G_VOID | G_DISCARD);
        }
      else if (ptr == (char *)p)
        return -1;
      else
        break;
    }

  return ptr - (char *)p;
}

static int
our_close (Vio *vio)
{
  if (vio->read != our_read)
    croak ("vio.read has unexpected content during unpatch - wtf?");

  if (vio->write != our_write)
    croak ("vio.write has unexpected content during unpatch - wtf?");

  if (vio->vioclose != our_close)
    croak ("vio.vioclose has unexpected content during unpatch - wtf?");

  SvREFCNT_dec (OURDATAPTR->corohandle);
  SvREFCNT_dec (OURDATAPTR->corohandle_sv);

  Safefree (OURDATAPTR);

  vio->read     = vio_read;
  vio->write    = vio_write;
  vio->vioclose = vio_close;

  vio->vioclose (vio);
}

MODULE = Coro::Mysql            PACKAGE = Coro::Mysql

BOOT:
{
  readable = get_cv ("Coro::Mysql::readable", 0);
  writable = get_cv ("Coro::Mysql::writable", 0);
}

PROTOTYPES: ENABLE

void
_patch (IV sock, int fd, SV *corohandle_sv, SV *corohandle)
        CODE:
{
        MYSQL *my = (MYSQL *)sock;
        Vio *vio = my->net.vio;
        ourdata *our;

        if (fd != my->net.fd)
          croak ("DBD::mysql fd and libmysql disagree - library mismatch, unsupported transport or API changes?");

        if (fd != vio->sd)
          croak ("DBD::mysql fd and vio-sd disagree - library mismatch, unsupported transport or API changes?");

        if (vio->vioclose != vio_close)
          croak ("vio.write has unexpected content - library mismatch, unsupported transport or API changes?");

        if (vio->write != vio_write)
          croak ("vio.write has unexpected content - library mismatch, unsupported transport or API changes?");

        if (vio->read != vio_read
            && vio->read != vio_read_buff)
          croak ("vio.read has unexpected content - library mismatch, unsupported transport or API changes?");

        Newz (0, our, 1, ourdata);
        our->magic = CoMy_MAGIC;
        our->corohandle_sv = newSVsv (corohandle_sv);
        our->corohandle    = newSVsv (corohandle);

        vio->desc [DESC_OFFSET - 1] = 0;
        OURDATAPTR = our;

        vio->vioclose = our_close;
        vio->write    = our_write;
        vio->read     = our_read;
}

Revision:	1.6
Committed:	Wed Sep 1 16:38:07 2010 UTC (13 years, 9 months ago) by root
Branch:	MAIN
CVS Tags:	rel-1_02
Changes since 1.5:	+2 -2 lines
Log Message:	1.02
#	User	Rev	Content
1	root	1.1	#include <sys/errno.h>
2			#include <unistd.h>
3			#include <fcntl.h>
4
5			#include <mysql.h>
6
7			#include "EXTERN.h"
8			#include "perl.h"
9			#include "XSUB.h"
10
11	root	1.3	#define IN_DESTRUCT PL_dirty
12
13	root	1.1	typedef U16 uint16;
14
15			/* cached function gv's */
16			static CV readable, writable;
17
18			#include "violite.h"
19
20			#define DESC_OFFSET 22
21
22			#define CoMy_MAGIC 0x436f4d79
23
24			typedef struct {
25			int magic;
26	root	1.5	SV corohandle_sv, corohandle;
27	root	1.1	int bufofs, bufcnt;
28			char buf[VIO_READ_BUFFER_SIZE];
29			} ourdata;
30
31			#define OURDATAPTR (((ourdata *)((vio)->desc + DESC_OFFSET)))
32
33			static int
34	root	1.6	our_read (Vio *vio, xgptr p, int len)
35	root	1.1	{
36			ourdata *our = OURDATAPTR;
37
38			if (!our->bufcnt)
39			{
40			int rd;
41			my_bool dummy;
42
43			vio->vioblocking (vio, 0, &dummy);
44
45			for (;;)
46			{
47			rd = recv (vio->sd, our->buf, sizeof (our->buf), 0);
48
49			if (rd >= 0 \|\| errno != EAGAIN)
50			break;
51
52			{
53			dSP;
54			PUSHMARK (SP);
55	root	1.5	XPUSHs (our->corohandle);
56	root	1.1	PUTBACK;
57			call_sv ((SV *)readable, G_VOID \| G_DISCARD);
58			}
59			}
60
61			if (rd <= 0)
62			return rd;
63
64			our->bufcnt = rd;
65			our->bufofs = 0;
66			}
67
68			if (our->bufcnt < len)
69			len = our->bufcnt;
70
71			memcpy (p, our->buf + our->bufofs, len);
72			our->bufofs += len;
73			our->bufcnt -= len;
74
75			return len;
76			}
77
78			static int
79	root	1.6	our_write (Vio *vio, const xgptr p, int len)
80	root	1.1	{
81			char ptr = (char )p;
82			my_bool dummy;
83
84			vio->vioblocking (vio, 0, &dummy);
85
86			while (len > 0)
87			{
88			int wr = send (vio->sd, ptr, len, 0);
89
90			if (wr > 0)
91			{
92			ptr += wr;
93			len -= wr;
94			}
95			else if (errno == EAGAIN)
96			{
97			dSP;
98			PUSHMARK (SP);
99	root	1.5	XPUSHs (OURDATAPTR->corohandle);
100	root	1.1	PUTBACK;
101			call_sv ((SV *)writable, G_VOID \| G_DISCARD);
102			}
103			else if (ptr == (char *)p)
104			return -1;
105			else
106			break;
107			}
108
109			return ptr - (char *)p;
110			}
111
112	root	1.5	static int
113			our_close (Vio *vio)
114			{
115			if (vio->read != our_read)
116			croak ("vio.read has unexpected content during unpatch - wtf?");
117
118			if (vio->write != our_write)
119			croak ("vio.write has unexpected content during unpatch - wtf?");
120
121			if (vio->vioclose != our_close)
122			croak ("vio.vioclose has unexpected content during unpatch - wtf?");
123
124			SvREFCNT_dec (OURDATAPTR->corohandle);
125			SvREFCNT_dec (OURDATAPTR->corohandle_sv);
126
127			Safefree (OURDATAPTR);
128
129			vio->read = vio_read;
130			vio->write = vio_write;
131			vio->vioclose = vio_close;
132
133			vio->vioclose (vio);
134			}
135
136	root	1.1	MODULE = Coro::Mysql PACKAGE = Coro::Mysql
137
138			BOOT:
139			{
140			readable = get_cv ("Coro::Mysql::readable", 0);
141			writable = get_cv ("Coro::Mysql::writable", 0);
142			}
143
144			PROTOTYPES: ENABLE
145
146			void
147	root	1.5	_patch (IV sock, int fd, SV corohandle_sv, SV corohandle)
148	root	1.1	CODE:
149			{
150			MYSQL my = (MYSQL )sock;
151			Vio *vio = my->net.vio;
152			ourdata *our;
153
154			if (fd != my->net.fd)
155			croak ("DBD::mysql fd and libmysql disagree - library mismatch, unsupported transport or API changes?");
156
157			if (fd != vio->sd)
158			croak ("DBD::mysql fd and vio-sd disagree - library mismatch, unsupported transport or API changes?");
159
160	root	1.5	if (vio->vioclose != vio_close)
161			croak ("vio.write has unexpected content - library mismatch, unsupported transport or API changes?");
162
163	root	1.1	if (vio->write != vio_write)
164			croak ("vio.write has unexpected content - library mismatch, unsupported transport or API changes?");
165
166	root	1.5	if (vio->read != vio_read
167			&& vio->read != vio_read_buff)
168	root	1.1	croak ("vio.read has unexpected content - library mismatch, unsupported transport or API changes?");
169
170			Newz (0, our, 1, ourdata);
171			our->magic = CoMy_MAGIC;
172	root	1.5	our->corohandle_sv = newSVsv (corohandle_sv);
173			our->corohandle = newSVsv (corohandle);
174	root	1.1
175			vio->desc [DESC_OFFSET - 1] = 0;
176			OURDATAPTR = our;
177
178	root	1.5	vio->vioclose = our_close;
179			vio->write = our_write;
180			vio->read = our_read;
181	root	1.1	}
182