#include "EXTERN.h" #include "perl.h" #include "XSUB.h" #define NEED_mg_findext #include "ppport.h" #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #include #define AUTOCREATE 1 #define OR_DIE 2 struct socketalarm; #include "SocketAlarm_util.h" #include "SocketAlarm_action.h" #include "pollfd_rbhash.h" #include "SocketAlarm_watcher.h" #define EVENT_SHUT 0x01 #define EVENT_EOF 0x02 #define EVENT_CLOSE 0x04 #define EVENT_IN 0x08 #define EVENT_PRI 0x10 #ifdef POLLRDHUP #define EVENT_DEFAULTS EVENT_SHUT #else #define EVENT_DEFAULTS (EVENT_SHUT|EVENT_EOF) #endif struct socketalarm { int list_ofs; // position within watch_list, initially -1 until activated int watch_fd; dev_t watch_fd_dev; ino_t watch_fd_ino; int event_mask; int action_count; SV *owner; AV *actions_av; // lazy-built bool unwaitable; // int cur_action; // used during execution struct timespec wake_ts; // struct action actions[]; }; static void socketalarm_exec_actions(struct socketalarm *sa); #include "SocketAlarm_util.c" #include "SocketAlarm_action.c" #include "SocketAlarm_watcher.c" struct socketalarm * socketalarm_new(int watch_fd, struct stat *statbuf, int event_mask, SV **action_spec, size_t spec_count) { size_t n_actions= 0, aux_len= 0, len_before_aux; struct socketalarm *self= NULL; parse_actions(action_spec, spec_count, NULL, &n_actions, NULL, &aux_len); // buffer needs aligned to pointer, which sizeof(struct action) is not guaranteed to be len_before_aux= sizeof(struct socketalarm) + n_actions * sizeof(struct action); len_before_aux += sizeof(void*)-1; len_before_aux &= ~(sizeof(void*)-1); self= (struct socketalarm *) safecalloc(1, len_before_aux + aux_len); // second call should succeed, because we gave it back it's own requested buffer sizes. // could fail if user did something evil like a tied scalar that changes length... if (!parse_actions(action_spec, spec_count, self->actions, &n_actions, ((char*)self) + len_before_aux, &aux_len)) croak("BUG: buffers not large enough for parse_actions"); // If user requests EVENT_SHUT and not available on the platform, warn and downgrade to EVENT_EOF if (!(EVENT_DEFAULTS & EVENT_SHUT) && (event_mask & EVENT_SHUT) && !(event_mask & EVENT_EOF)) { SV *warned= get_sv("IO::SocketAlarm::_warned_EVENT_SHUT_unavail", GV_ADD); if (!SvTRUE(warned)) { warn("IO::SocketAlarm EVENT_SHUT is not available on this platform, using EVENT_EOF instead"); sv_setsv(warned, &PL_sv_yes); } event_mask |= EVENT_EOF; } self->watch_fd= watch_fd; self->watch_fd_dev= statbuf->st_dev; self->watch_fd_ino= statbuf->st_ino; self->event_mask= event_mask; self->actions_av= NULL; self->action_count= n_actions; self->list_ofs= -1; // initially not in the watch list self->owner= NULL; return self; } void socketalarm_exec_actions(struct socketalarm *self) { bool resume= self->cur_action >= 0; struct timespec now_ts= { 0, -1 }; if (!resume) self->cur_action= 0; while (self->cur_action < self->action_count) { if (!execute_action(self->actions + self->cur_action, resume, &now_ts, self)) break; resume= false; ++self->cur_action; } } static void socketalarm__build_actions(struct socketalarm *self) { Size_t i; if (!self->actions_av) { self->actions_av= newAV(); av_extend(self->actions_av, self->action_count-1); for (i= 0; i < self->action_count; i++) { AV *act_spec= newAV(); inflate_action(self->actions+i, act_spec); SvREADONLY_on((SV*) act_spec); av_push(self->actions_av, newRV_noinc((SV*) act_spec)); } SvREADONLY_on((SV*) self->actions_av); } } void socketalarm_free(struct socketalarm *sa) { // Must remove the socketalarm from the active list, if present watch_list_remove(sa); // Release reference to lazy-built action AV if (sa->actions_av) SvREFCNT_dec((SV*) sa->actions_av); // was allocated as one chunk Safefree(sa); } /* Return an SV array of an AV. * Returns NULL if it wasn't an AV or arrayref. */ static SV** unwrap_array(SV *array, SSize_t *len_out) { AV *av; SV **vec; SSize_t n; if (array && SvTYPE(array) == SVt_PVAV) av= (AV*) array; else if (array && SvROK(array) && SvTYPE(SvRV(array)) == SVt_PVAV) av= (AV*) SvRV(array); else return NULL; n= av_count(av); vec= AvARRAY(av); /* tied arrays and non-allocated empty arrays return NULL */ if (!vec) { if (n == 0) /* don't return a NULL for an empty array, but doesn't need to be a real pointer */ vec= (SV**) 8; else { /* in case of a tied array, extract the elements into a temporary buffer */ SSize_t i; Newx(vec, n, SV*); SAVEFREEPV(vec); for (i= 0; i < n; i++) { SV **el= av_fetch(av, i, 0); vec[i]= el? *el : NULL; } } } if (len_out) *len_out= n; return vec; } /*------------------------------------------------------------------------------------ * Definitions of Perl MAGIC that attach C structs to Perl SVs */ // destructor for Watch objects static int socketalarm_magic_free(pTHX_ SV* sv, MAGIC* mg) { if (mg->mg_ptr) { socketalarm_free((struct socketalarm*) mg->mg_ptr); mg->mg_ptr= NULL; } return 0; // ignored anyway } #ifdef USE_ITHREADS static int socketalarm_magic_dup(pTHX_ MAGIC *mg, CLONE_PARAMS *param) { croak("This object cannot be shared between threads"); return 0; }; #else #define socketalarm_magic_dup 0 #endif // magic table for Watch objects static MGVTBL socketalarm_magic_vt= { 0, /* get */ 0, /* write */ 0, /* length */ 0, /* clear */ socketalarm_magic_free, 0, /* copy */ socketalarm_magic_dup #ifdef MGf_LOCAL ,0 #endif }; // Return the socketalarm that was attached to a perl Watch object via MAGIC. // The 'obj' should be a reference to a blessed magical SV. static struct socketalarm* get_magic_socketalarm(SV *obj, int flags) { SV *sv; MAGIC* magic; if (!sv_isobject(obj)) { if (flags & OR_DIE) croak("Not an object"); return NULL; } sv= SvRV(obj); if (SvMAGICAL(sv)) { /* Iterate magic attached to this scalar, looking for one with our vtable */ if ((magic= mg_findext(sv, PERL_MAGIC_ext, &socketalarm_magic_vt))) /* If found, the mg_ptr points to the fields structure. */ return (struct socketalarm*) magic->mg_ptr; } if (flags & OR_DIE) croak("Object lacks 'struct TreeRBXS_item' magic"); return NULL; } static void attach_magic_socketalarm(SV *obj_inner_sv, struct socketalarm *sa) { MAGIC *magic; if (sa->owner) croak("BUG: already attached to perl object"); sa->owner= obj_inner_sv; magic= sv_magicext((SV*) sa->owner, NULL, PERL_MAGIC_ext, &socketalarm_magic_vt, (const char*) sa, 0); #ifdef USE_ITHREADS magic->mg_flags |= MGf_DUP; #else (void)magic; // suppress 'unused' warning #endif } // Return existing Watch object, or create a new one. // Returned SV has a non-mortal refcount, which is what the typemap // wants for returning a "struct socketalarm*" to perl-land static SV* wrap_socketalarm(struct socketalarm *sa) { SV *obj; HV *hv; // Since this is used in typemap, handle NULL gracefully if (!sa) return &PL_sv_undef; // If there is already a node object, return a new reference to it. if (sa->owner) return newRV_inc((SV*) sa->owner); // else create a node object hv= newHV(); obj= newRV_noinc((SV*) hv); sv_bless(obj, gv_stashpv("IO::SocketAlarm", GV_ADD)); attach_magic_socketalarm((SV*) hv, sa); return obj; } #define EXPORT_ENUM(x) newCONSTSUB(stash, #x, new_enum_dualvar(aTHX_ x, newSVpvs_share(#x))) static SV * new_enum_dualvar(pTHX_ IV ival, SV *name) { SvUPGRADE(name, SVt_PVNV); SvIV_set(name, ival); SvIOK_on(name); SvREADONLY_on(name); return name; } /*------------------------------------------------------------------------------------ * Perl API */ MODULE = IO::SocketAlarm PACKAGE = IO::SocketAlarm void _init_socketalarm(self, sock_sv, eventmask_sv, actions_sv) SV *self SV *sock_sv SV *eventmask_sv SV *actions_sv INIT: int sock_fd= fileno_from_sv(sock_sv); int eventmask= EVENT_DEFAULTS; struct stat statbuf; struct socketalarm *sa; SV **action_list= NULL; SSize_t n_actions= 0; PPCODE: if (!sv_isobject(self)) croak("Not an object"); if ((sa= get_magic_socketalarm(self, 0))) croak("Already initialized"); if (!(sock_fd >= 0 && fstat(sock_fd, &statbuf) == 0 && S_ISSOCK(statbuf.st_mode))) croak("Not an open socket"); if (eventmask_sv && SvOK(eventmask_sv)) eventmask= SvIV(eventmask_sv); if (actions_sv && SvOK(actions_sv)) { action_list= unwrap_array(actions_sv, &n_actions); if (!action_list) croak("Actions must be an arrayref (or undefined)"); } sa= socketalarm_new(sock_fd, &statbuf, eventmask, action_list, n_actions); attach_magic_socketalarm(SvRV(self), sa); XSRETURN(1); // return $self int socket(alarm) struct socketalarm *alarm CODE: RETVAL= alarm->watch_fd; OUTPUT: RETVAL int events(alarm) struct socketalarm *alarm CODE: RETVAL= alarm->event_mask; OUTPUT: RETVAL void actions(alarm) struct socketalarm *alarm PPCODE: if (!alarm->actions_av); socketalarm__build_actions(alarm); ST(0)= sv_2mortal(newRV_inc((SV*) alarm->actions_av)); XSRETURN(1); int action_count(alarm) struct socketalarm *alarm CODE: RETVAL= alarm->action_count; OUTPUT: RETVAL int cur_action(alarm) struct socketalarm *alarm CODE: watch_list_item_get_status(alarm, &RETVAL); OUTPUT: RETVAL bool start(alarm) struct socketalarm *alarm CODE: RETVAL= watch_list_add(alarm); OUTPUT: RETVAL bool cancel(alarm) struct socketalarm *alarm CODE: RETVAL= watch_list_remove(alarm); OUTPUT: RETVAL SV* stringify(alarm) struct socketalarm *alarm INIT: SV *out= sv_2mortal(newSVpvn("",0)); Size_t i; CODE: sv_catpvf(out, "watch fd: %d\n", alarm->watch_fd); sv_catpvf(out, "event mask:%s%s\n", alarm->event_mask & EVENT_SHUT? " SHUT":"", alarm->event_mask & EVENT_CLOSE? " CLOSE":"" ); sv_catpv(out, "actions:\n"); for (i= 0; i < alarm->action_count; i++) { char buf[256]; snprint_action(buf, sizeof(buf), alarm->actions+i); sv_catpvf(out, "%4d: %s\n", (int)i, buf); } SvREFCNT_inc(out); RETVAL= out; OUTPUT: RETVAL void _terminate_all() PPCODE: shutdown_watch_thread(); MODULE = IO::SocketAlarm PACKAGE = IO::SocketAlarm::Util struct socketalarm * socketalarm(sock_sv, ...) SV *sock_sv INIT: int sock_fd= fileno_from_sv(sock_sv); int eventmask= EVENT_DEFAULTS; int action_ofs= 1; struct stat statbuf; CODE: if (!(sock_fd >= 0 && fstat(sock_fd, &statbuf) == 0 && S_ISSOCK(statbuf.st_mode))) croak("Not an open socket"); if (items > 1) { // must either be a scalar, a scalar followed by actions specs, or action specs if (SvOK(ST(1)) && looks_like_number(ST(1))) { eventmask= SvIV(ST(1)); action_ofs++; } } RETVAL= socketalarm_new(sock_fd, &statbuf, eventmask, &(ST(action_ofs)), items - action_ofs); watch_list_add(RETVAL); OUTPUT: RETVAL bool is_socket(fd_sv) SV *fd_sv INIT: int fd= fileno_from_sv(fd_sv); struct stat statbuf; CODE: RETVAL= fd >= 0 && fstat(fd, &statbuf) == 0 && S_ISSOCK(statbuf.st_mode); OUTPUT: RETVAL SV * get_fd_table_str(max_fd=1024) int max_fd INIT: SV *out= newSVpvn("",0); size_t avail= 0, needed= 1023; CODE: // FD status could change between calls, changing the length requirement, so loop. // 'avail' count includes the NUL byte, and 'needed' does not. while (avail <= needed) { sv_grow(out, needed+1); avail= needed+1; needed= snprint_fd_table(SvPVX(out), avail, max_fd); } SvCUR_set(out, needed); RETVAL= out; OUTPUT: RETVAL # For unit test purposes only, export _poll that polls on a single file # descriptor to verify the statuses for sockets in various states. void _poll(fd, events, timeout) int fd SV *events; int timeout; INIT: int ret; struct pollfd pollbuf; PPCODE: pollbuf.fd= fd; pollbuf.events= SvIV(events); pollbuf.revents= 0; ret= poll(&pollbuf, 1, timeout); EXTEND(SP, 2); PUSHs(sv_2mortal(newSViv(ret))); PUSHs(sv_2mortal(newSViv(ret < 0? errno : ret > 0? pollbuf.revents : 0))); #----------------------------------------------------------------------------- # Constants # BOOT: HV* stash= gv_stashpvn("IO::SocketAlarm::Util", 21, GV_ADD); EXPORT_ENUM(EVENT_SHUT); EXPORT_ENUM(EVENT_EOF); EXPORT_ENUM(EVENT_IN); EXPORT_ENUM(EVENT_PRI); EXPORT_ENUM(EVENT_CLOSE); EXPORT_ENUM(POLLIN); EXPORT_ENUM(POLLOUT); EXPORT_ENUM(POLLPRI); EXPORT_ENUM(POLLERR); EXPORT_ENUM(POLLHUP); #ifdef POLLRDHUP EXPORT_ENUM(POLLRDHUP); #endif EXPORT_ENUM(POLLNVAL); PROTOTYPES: DISABLE