diff options
Diffstat (limited to 'erts/emulator')
-rw-r--r-- | erts/emulator/nifs/common/socket_nif.c | 798 |
1 files changed, 762 insertions, 36 deletions
diff --git a/erts/emulator/nifs/common/socket_nif.c b/erts/emulator/nifs/common/socket_nif.c index 6564c3c82f..c48d6eab00 100644 --- a/erts/emulator/nifs/common/socket_nif.c +++ b/erts/emulator/nifs/common/socket_nif.c @@ -104,6 +104,14 @@ #undef WANT_NONBLOCKING #include "sys.h" + + + +/* AND HERE WE MAY HAVE A BUNCH OF DEFINES....SEE INET DRIVER.... */ + + + + #else /* !__WIN32__ */ #include <sys/time.h> @@ -297,7 +305,7 @@ static void (*esock_sctp_freepaddrs)(struct sockaddr *addrs) = NULL; #endif #include "sys.h" -#endif +#endif /* !__WIN32__ */ #include <erl_nif.h> @@ -731,6 +739,7 @@ typedef struct { SocketAddress remote; unsigned int addrLen; + ErlNifEnv* env; /* +++ Controller (owner) process +++ */ ErlNifPid ctrlPid; @@ -771,7 +780,7 @@ typedef struct { size_t rBufSz; // Read buffer size (when data length = 0 is specified) size_t rCtrlSz; // Read control buffer size size_t wCtrlSz; // Write control buffer size - BOOLEAN_T iow; // Inform On Wrap + BOOLEAN_T iow; // Inform On (counter) Wrap BOOLEAN_T dbg; /* +++ Close stuff +++ */ @@ -909,11 +918,9 @@ static ERL_NIF_TERM nif_finalize_connection(ErlNifEnv* env, static ERL_NIF_TERM nif_finalize_close(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]); -/* static ERL_NIF_TERM nif_cancel(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]); -*/ static ERL_NIF_TERM nopen(ErlNifEnv* env, @@ -977,7 +984,6 @@ static ERL_NIF_TERM nclose(ErlNifEnv* env, static ERL_NIF_TERM nshutdown(ErlNifEnv* env, SocketDescriptor* descP, int how); - static ERL_NIF_TERM nsetopt(ErlNifEnv* env, SocketDescriptor* descP, BOOLEAN_T isEncoded, @@ -1817,6 +1823,43 @@ static ERL_NIF_TERM nsockname(ErlNifEnv* env, SocketDescriptor* descP); static ERL_NIF_TERM npeername(ErlNifEnv* env, SocketDescriptor* descP); +static ERL_NIF_TERM ncancel(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM op, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_connect(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_accept(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_accept_current(ErlNifEnv* env, + SocketDescriptor* descP); +static ERL_NIF_TERM ncancel_accept_waiting(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_send(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_send_current(ErlNifEnv* env, + SocketDescriptor* descP); +static ERL_NIF_TERM ncancel_send_waiting(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_recv(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_read_select(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_write_select(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef); +static ERL_NIF_TERM ncancel_mode_select(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef, + int smode, + int rmode); static ERL_NIF_TERM nsetopt_str_opt(ErlNifEnv* env, SocketDescriptor* descP, @@ -1858,6 +1901,10 @@ static ERL_NIF_TERM ngetopt_timeval_opt(ErlNifEnv* env, int level, int opt); +static BOOLEAN_T send_check_writer(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM ref, + ERL_NIF_TERM* checkResult); static ERL_NIF_TERM send_check_result(ErlNifEnv* env, SocketDescriptor* descP, ssize_t written, @@ -2083,6 +2130,25 @@ static BOOLEAN_T acceptor_pop(ErlNifEnv* env, ErlNifPid* pid, ErlNifMonitor* mon, ERL_NIF_TERM* ref); +static BOOLEAN_T acceptor_unqueue(ErlNifEnv* env, + SocketDescriptor* descP, + const ErlNifPid* pid); + +static BOOLEAN_T writer_search4pid(ErlNifEnv* env, + SocketDescriptor* descP, + ErlNifPid* pid); +static ERL_NIF_TERM writer_push(ErlNifEnv* env, + SocketDescriptor* descP, + ErlNifPid pid, + ERL_NIF_TERM ref); +static BOOLEAN_T writer_pop(ErlNifEnv* env, + SocketDescriptor* descP, + ErlNifPid* pid, + ErlNifMonitor* mon, + ERL_NIF_TERM* ref); +static BOOLEAN_T writer_unqueue(ErlNifEnv* env, + SocketDescriptor* descP, + const ErlNifPid* pid); static BOOLEAN_T qsearch4pid(ErlNifEnv* env, SocketRequestQueue* q, @@ -2235,8 +2301,10 @@ static char str_exsend[] = "exsend"; // failed send /* *** "Global" Atoms *** */ +ERL_NIF_TERM esock_atom_accept; ERL_NIF_TERM esock_atom_addr; ERL_NIF_TERM esock_atom_any; +ERL_NIF_TERM esock_atom_connect; ERL_NIF_TERM esock_atom_credentials; ERL_NIF_TERM esock_atom_ctrl; ERL_NIF_TERM esock_atom_ctrunc; @@ -2261,6 +2329,7 @@ ERL_NIF_TERM esock_atom_local; ERL_NIF_TERM esock_atom_loopback; ERL_NIF_TERM esock_atom_lowdelay; ERL_NIF_TERM esock_atom_mincost; +ERL_NIF_TERM esock_atom_not_found; ERL_NIF_TERM esock_atom_ok; ERL_NIF_TERM esock_atom_oob; ERL_NIF_TERM esock_atom_origdstaddr; @@ -2270,11 +2339,18 @@ ERL_NIF_TERM esock_atom_port; ERL_NIF_TERM esock_atom_protocol; ERL_NIF_TERM esock_atom_raw; ERL_NIF_TERM esock_atom_rdm; -ERL_NIF_TERM esock_atom_rights; +ERL_NIF_TERM esock_atom_recv; +ERL_NIF_TERM esock_atom_recvfrom; +ERL_NIF_TERM esock_atom_recvmsg; ERL_NIF_TERM esock_atom_reliability; +ERL_NIF_TERM esock_atom_rights; ERL_NIF_TERM esock_atom_scope_id; ERL_NIF_TERM esock_atom_sctp; ERL_NIF_TERM esock_atom_sec; +ERL_NIF_TERM esock_atom_select_sent; +ERL_NIF_TERM esock_atom_send; +ERL_NIF_TERM esock_atom_sendmsg; +ERL_NIF_TERM esock_atom_sendto; ERL_NIF_TERM esock_atom_seqpacket; ERL_NIF_TERM esock_atom_socket; ERL_NIF_TERM esock_atom_spec_dst; @@ -3222,7 +3298,8 @@ ERL_NIF_TERM naccept_listening(ErlNifEnv* env, &descP->currentAcceptor.mon) > 0) return esock_make_error(env, atom_exmon); - descP->currentAcceptor.ref = ref; + descP->currentAcceptor.ref = enif_make_copy(descP->env, ref); + descP->currentAcceptorP = &descP->currentAcceptor; SELECT(env, descP->sock, @@ -3320,7 +3397,8 @@ ERL_NIF_TERM naccept_listening(ErlNifEnv* env, /* *** naccept_accepting *** * We have an active acceptor and possibly acceptors waiting in queue. - * At the moment the queue is *not* implemented. + * If the pid of the calling process is not the pid of the "current process", + * push the requester onto the queue. */ static ERL_NIF_TERM naccept_accepting(ErlNifEnv* env, @@ -3347,11 +3425,8 @@ ERL_NIF_TERM naccept_accepting(ErlNifEnv* env, "\r\n", caller, descP->currentAcceptor.pid) ); if (!compare_pids(env, &descP->currentAcceptor.pid, &caller)) { - /* This will have to do until we implement the queue. - * When we have the queue, we should simply push this request, - * and instead return with eagain (the caller will then wait - * for the select message). - */ + + /* Not the "current acceptor", so (maybe) push onto queue */ SSDBG( descP, ("SOCKET", "naccept_accepting -> not (active) acceptor\r\n") ); @@ -3411,6 +3486,8 @@ ERL_NIF_TERM naccept_accepting(ErlNifEnv* env, SSDBG( descP, ("SOCKET", "naccept_accepting -> accept success\r\n") ); + DEMONP(env, descP, &descP->currentAcceptor.mon); + if ((accEvent = sock_create_event(accSock)) == INVALID_EVENT) { save_errno = sock_errno(); while ((sock_close(accSock) == INVALID_SOCKET) && @@ -3451,10 +3528,7 @@ ERL_NIF_TERM naccept_accepting(ErlNifEnv* env, accDescP->state = SOCKET_STATE_CONNECTED; - /* Here we should have the test if we have something in the queue. - * And if so, pop it and copy the (waiting) acceptor, and then - * make a new select with that info). - */ + /* Check if there are waiting acceptors (popping the acceptor queue) */ if (acceptor_pop(env, descP, &descP->currentAcceptor.pid, @@ -3559,12 +3633,12 @@ ERL_NIF_TERM nif_send(ErlNifEnv* env, } -/* What do we do when another process tries to write - * when the current writer has a select already waiting? - * Queue it? And what about simultaneous read and write? - * Queue up all operations towards the socket? +/* *** nsend *** * - * We (may) need a currentOp field and an ops queue field. + * Do the actual send. + * Do some initial writer checks, do the actual send and then + * analyze the result. If we are done, another writer may be + * scheduled (if there is one in the writer queue). */ static ERL_NIF_TERM nsend(ErlNifEnv* env, @@ -3573,12 +3647,17 @@ ERL_NIF_TERM nsend(ErlNifEnv* env, ErlNifBinary* sndDataP, int flags) { - int save_errno; - ssize_t written; + int save_errno; + ssize_t written; + ERL_NIF_TERM writerCheck; if (!descP->isWritable) return enif_make_badarg(env); + /* Check if there is already a current writer and if its us */ + if (!send_check_writer(env, descP, sendRef, &writerCheck)) + return writerCheck; + /* We ignore the wrap for the moment. * Maybe we should issue a wrap-message to controlling process... */ @@ -3597,6 +3676,7 @@ ERL_NIF_TERM nsend(ErlNifEnv* env, } + /* ---------------------------------------------------------------------- * nif_sendto * @@ -3662,9 +3742,13 @@ ERL_NIF_TERM nif_sendto(ErlNifEnv* env, &remoteAddrLen)) != NULL) return esock_make_error_str(env, xres); + MLOCK(descP->writeMtx); + res = nsendto(env, descP, sendRef, &sndData, flags, &remoteAddr, remoteAddrLen); + MUNLOCK(descP->writeMtx); + SGDBG( ("SOCKET", "nif_sendto -> done with result: " "\r\n %T" "\r\n", res) ); @@ -3682,12 +3766,17 @@ ERL_NIF_TERM nsendto(ErlNifEnv* env, SocketAddress* toAddrP, unsigned int toAddrLen) { - int save_errno; - ssize_t written; + int save_errno; + ssize_t written; + ERL_NIF_TERM writerCheck; if (!descP->isWritable) return enif_make_badarg(env); + /* Check if there is already a current writer and if its us */ + if (!send_check_writer(env, descP, sendRef, &writerCheck)) + return writerCheck; + /* We ignore the wrap for the moment. * Maybe we should issue a wrap-message to controlling process... */ @@ -3763,8 +3852,12 @@ ERL_NIF_TERM nif_sendmsg(ErlNifEnv* env, if (!esendflags2sendflags(eflags, &flags)) return esock_make_error(env, esock_atom_einval); + MLOCK(descP->writeMtx); + res = nsendmsg(env, descP, sendRef, eMsgHdr, flags); + MUNLOCK(descP->writeMtx); + SSDBG( descP, ("SOCKET", "nif_sendmsg -> done with result: " "\r\n %T" @@ -3791,12 +3884,16 @@ ERL_NIF_TERM nsendmsg(ErlNifEnv* env, size_t ctrlBufLen, ctrlBufUsed; int save_errno; ssize_t written, dataSize; + ERL_NIF_TERM writerCheck; char* xres; if (!descP->isWritable) return enif_make_badarg(env); - + /* Check if there is already a current writer and if its us */ + if (!send_check_writer(env, descP, sendRef, &writerCheck)) + return writerCheck; + /* Depending on if we are *connected* or not, we require * different things in the msghdr map. */ @@ -7175,6 +7272,7 @@ ERL_NIF_TERM nsetopt_lvl_sctp_associnfo(ErlNifEnv* env, int res; size_t sz; unsigned int tmp; + int32_t tmpAssocId; SSDBG( descP, ("SOCKET", "nsetopt_lvl_sctp_associnfo -> entry with" @@ -7213,8 +7311,12 @@ ERL_NIF_TERM nsetopt_lvl_sctp_associnfo(ErlNifEnv* env, SSDBG( descP, ("SOCKET", "nsetopt_lvl_sctp_associnfo -> decode attributes\r\n") ); - if (!GET_UINT(env, eAssocId, &assocParams.sasoc_assoc_id)) + /* On some platforms the assoc id is typed as an unsigned integer (uint32) + * So, to avoid warnings there, we always make an explicit cast... + */ + if (!GET_INT(env, eAssocId, &tmpAssocId)) return esock_make_error(env, esock_atom_einval); + assocParams.sasoc_assoc_id = (typeof(assocParams.sasoc_assoc_id)) tmpAssocId; /* * We should really make sure this is ok in erlang (to ensure that @@ -7509,6 +7611,7 @@ ERL_NIF_TERM nsetopt_lvl_sctp_rtoinfo(ErlNifEnv* env, struct sctp_rtoinfo rtoInfo; int res; size_t sz; + int32_t tmpAssocId; SSDBG( descP, ("SOCKET", "nsetopt_lvl_sctp_rtoinfo -> entry with" @@ -7541,8 +7644,12 @@ ERL_NIF_TERM nsetopt_lvl_sctp_rtoinfo(ErlNifEnv* env, SSDBG( descP, ("SOCKET", "nsetopt_lvl_sctp_rtoinfo -> decode attributes\r\n") ); - if (!GET_UINT(env, eAssocId, &rtoInfo.srto_assoc_id)) + /* On some platforms the assoc id is typed as an unsigned integer (uint32) + * So, to avoid warnings there, we always make an explicit cast... + */ + if (!GET_INT(env, eAssocId, &tmpAssocId)) return esock_make_error(env, esock_atom_einval); + rtoInfo.srto_assoc_id = (typeof(rtoInfo.srto_assoc_id)) tmpAssocId; if (!GET_UINT(env, eInitial, &rtoInfo.srto_initial)) return esock_make_error(env, esock_atom_einval); @@ -10414,10 +10521,471 @@ ERL_NIF_TERM npeername(ErlNifEnv* env, /* ---------------------------------------------------------------------- + * nif_cancel + * + * Description: + * Cancel a previous select! + * + * Arguments: + * Socket (ref) - Points to the socket descriptor. + * Operation (atom) - What kind of operation (accept, send, ...) is to be cancelled + * Ref (ref) - Unique id for the operation + */ +static +ERL_NIF_TERM nif_cancel(ErlNifEnv* env, + int argc, + const ERL_NIF_TERM argv[]) +{ + SocketDescriptor* descP; + ERL_NIF_TERM op, opRef, result; + + SGDBG( ("SOCKET", "nif_cancel -> entry with argc: %d\r\n", argc) ); + + /* Extract arguments and perform preliminary validation */ + + if ((argc != 3) || + !enif_get_resource(env, argv[0], sockets, (void**) &descP)) { + return enif_make_badarg(env); + } + op = argv[1]; + opRef = argv[2]; + + SSDBG( descP, + ("SOCKET", "nif_cancel -> args when sock = %d:" + "\r\n op: %T" + "\r\n opRef: %T" + "\r\n", descP->sock, op, opRef) ); + + result = ncancel(env, descP, op, opRef); + + SSDBG( descP, + ("SOCKET", "nif_cancel -> done with result: " + "\r\n %T" + "\r\n", result) ); + + return result; + +} + + +static +ERL_NIF_TERM ncancel(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM op, + ERL_NIF_TERM opRef) +{ + /* <KOLLA> + * + * Do we really need all these variants? Should it not be enough with: + * + * connect | accept | send | recv + * + * </KOLLA> + */ + if (COMPARE(op, esock_atom_connect) == 0) { + return ncancel_connect(env, descP, opRef); + } else if (COMPARE(op, esock_atom_accept) == 0) { + return ncancel_accept(env, descP, opRef); + } else if (COMPARE(op, esock_atom_send) == 0) { + return ncancel_send(env, descP, opRef); + } else if (COMPARE(op, esock_atom_sendto) == 0) { + return ncancel_send(env, descP, opRef); + } else if (COMPARE(op, esock_atom_sendmsg) == 0) { + return ncancel_send(env, descP, opRef); + } else if (COMPARE(op, esock_atom_recv) == 0) { + return ncancel_recv(env, descP, opRef); + } else if (COMPARE(op, esock_atom_recvfrom) == 0) { + return ncancel_recv(env, descP, opRef); + } else if (COMPARE(op, esock_atom_recvmsg) == 0) { + return ncancel_recv(env, descP, opRef); + } else { + return esock_make_error(env, esock_atom_einval); + } +} + + + +/* *** ncancel_connect *** + * + * + */ +static +ERL_NIF_TERM ncancel_connect(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + return ncancel_write_select(env, descP, opRef); +} + + +/* *** ncancel_accept *** + * + * We have two different cases: + * *) Its the current acceptor + * Cancel the select! + * We need to activate one of the waiting acceptors. + * *) Its one of the acceptors ("waiting") in the queue + * Simply remove the acceptor from the queue. + * + */ +static +ERL_NIF_TERM ncancel_accept(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + ERL_NIF_TERM res; + + SSDBG( descP, + ("SOCKET", "ncancel_accept -> entry with" + "\r\n opRef: %T" + "\r\n %s" + "\r\n", opRef, + ((descP->currentAcceptorP == NULL) ? "without acceptor" : "with acceptor")) ); + + MLOCK(descP->accMtx); + + if (descP->currentAcceptorP != NULL) { + if (COMPARE(opRef, descP->currentAcceptor.ref) == 0) { + res = ncancel_accept_current(env, descP); + } else { + res = ncancel_accept_waiting(env, descP, opRef); + } + } else { + /* Or badarg? */ + res = esock_make_error(env, esock_atom_einval); + } + + MUNLOCK(descP->accMtx); + + SSDBG( descP, + ("SOCKET", "ncancel_accept -> done with result:" + "\r\n %T" + "\r\n", res) ); + + return res; +} + + +/* The current acceptor process has an ongoing select we first must + * cancel. Then we must re-activate the "first" (the first + * in the acceptor queue). + */ +static +ERL_NIF_TERM ncancel_accept_current(ErlNifEnv* env, + SocketDescriptor* descP) +{ + ERL_NIF_TERM res; + + SSDBG( descP, ("SOCKET", "ncancel_accept_current -> entry\r\n") ); + + res = ncancel_read_select(env, descP, descP->currentAcceptor.ref); + + SSDBG( descP, ("SOCKET", "ncancel_accept_current -> cancel res: %T\r\n", res) ); + + if (acceptor_pop(env, descP, + &descP->currentAcceptor.pid, + &descP->currentAcceptor.mon, + &descP->currentAcceptor.ref)) { + + /* There was another one */ + + SSDBG( descP, ("SOCKET", "ncancel_accept_current -> new (active) acceptor: " + "\r\n pid: %T" + "\r\n ref: %T" + "\r\n", + descP->currentAcceptor.pid, + descP->currentAcceptor.ref) ); + + SELECT(env, + descP->sock, + (ERL_NIF_SELECT_READ), + descP, &descP->currentAcceptor.pid, descP->currentAcceptor.ref); + + } else { + SSDBG( descP, ("SOCKET", "ncancel_accept_current -> no more acceptors\r\n") ); + descP->currentAcceptorP = NULL; + descP->state = SOCKET_STATE_LISTENING; + } + + SSDBG( descP, ("SOCKET", "ncancel_accept_current -> done with result:" + "\r\n %T" + "\r\n", res) ); + + return res; +} + + +/* These processes have not performed a select, so we can simply + * remove them from the acceptor queue. + */ +static +ERL_NIF_TERM ncancel_accept_waiting(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + ErlNifPid caller; + + if (enif_self(env, &caller) == NULL) + return esock_make_error(env, atom_exself); + + /* unqueue request from (acceptor) queue */ + + if (acceptor_unqueue(env, descP, &caller)) { + return esock_atom_ok; + } else { + /* Race? */ + return esock_make_error(env, esock_atom_not_found); + } +} + + + +/* *** ncancel_send *** + * + * Cancel a send operation. + * Its either the current writer or one of the waiting writers. + */ +static +ERL_NIF_TERM ncancel_send(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + ERL_NIF_TERM res; + + SSDBG( descP, + ("SOCKET", "ncancel_send -> entry with" + "\r\n opRef: %T" + "\r\n %s" + "\r\n", opRef, + ((descP->currentWriterP == NULL) ? "without writer" : "with writer")) ); + + MLOCK(descP->writeMtx); + + if (descP->currentWriterP != NULL) { + if (COMPARE(opRef, descP->currentWriter.ref) == 0) { + res = ncancel_send_current(env, descP); + } else { + res = ncancel_send_waiting(env, descP, opRef); + } + } else { + /* Or badarg? */ + res = esock_make_error(env, esock_atom_einval); + } + + MUNLOCK(descP->writeMtx); + + SSDBG( descP, + ("SOCKET", "ncancel_send -> done with result:" + "\r\n %T" + "\r\n", res) ); + + return res; +} + + + +/* The current writer process has an ongoing select we first must + * cancel. Then we must re-activate the "first" (the first + * in the writer queue). + */ +static +ERL_NIF_TERM ncancel_send_current(ErlNifEnv* env, + SocketDescriptor* descP) +{ + ERL_NIF_TERM res; + + SSDBG( descP, ("SOCKET", "ncancel_send_current -> entry\r\n") ); + + res = ncancel_write_select(env, descP, descP->currentWriter.ref); + + SSDBG( descP, ("SOCKET", "ncancel_send_current -> cancel res: %T\r\n", res) ); + + if (writer_pop(env, descP, + &descP->currentWriter.pid, + &descP->currentWriter.mon, + &descP->currentWriter.ref)) { + + /* There was another one */ + + SSDBG( descP, ("SOCKET", "ncancel_send_current -> new (active) writer: " + "\r\n pid: %T" + "\r\n ref: %T" + "\r\n", + descP->currentWriter.pid, + descP->currentWriter.ref) ); + + SELECT(env, + descP->sock, + (ERL_NIF_SELECT_WRITE), + descP, &descP->currentWriter.pid, descP->currentWriter.ref); + + } else { + SSDBG( descP, ("SOCKET", "ncancel_send_current -> no more writers\r\n") ); + descP->currentWriterP = NULL; + } + + SSDBG( descP, ("SOCKET", "ncancel_send_current -> done with result:" + "\r\n %T" + "\r\n", res) ); + + return res; +} + + +/* These processes have not performed a select, so we can simply + * remove them from the writer queue. + */ +static +ERL_NIF_TERM ncancel_send_waiting(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + ErlNifPid caller; + + if (enif_self(env, &caller) == NULL) + return esock_make_error(env, atom_exself); + + /* unqueue request from (writer) queue */ + + if (writer_unqueue(env, descP, &caller)) { + return esock_atom_ok; + } else { + /* Race? */ + return esock_make_error(env, esock_atom_not_found); + } +} + + + +/* *** ncancel_recv *** + * + * + */ +static +ERL_NIF_TERM ncancel_recv(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + return esock_make_error(env, esock_atom_einval); +} + + + +static +ERL_NIF_TERM ncancel_read_select(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + return ncancel_mode_select(env, descP, opRef, + ERL_NIF_SELECT_READ, + ERL_NIF_SELECT_READ_CANCELLED); +} + + +static +ERL_NIF_TERM ncancel_write_select(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef) +{ + return ncancel_mode_select(env, descP, opRef, + ERL_NIF_SELECT_WRITE, + ERL_NIF_SELECT_WRITE_CANCELLED); +} + + +static +ERL_NIF_TERM ncancel_mode_select(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM opRef, + int smode, + int rmode) +{ + int selectRes = enif_select(env, descP->sock, + (ERL_NIF_SELECT_CANCEL | smode), + descP, NULL, opRef); + + if (selectRes & rmode) { + /* Was cancelled */ + return esock_atom_ok; + } else if (selectRes > 0) { + /* Has already sent the message */ + return esock_make_error(env, esock_atom_select_sent); + } else { + /* Stopped? */ + SSDBG( descP, ("SOCKET", "ncancel_mode_select -> failed: %d (0x%lX)" + "\r\n", selectRes, selectRes) ); + return esock_make_error(env, esock_atom_einval); + } + +} + + + +/* ---------------------------------------------------------------------- * U t i l i t y F u n c t i o n s * ---------------------------------------------------------------------- */ +/* *** send_check_writer *** + * + * Checks if we have a current writer and if that is us. If not, then we must + * be made to wait for our turn. This is done by pushing us unto the writer queue. + */ +static +BOOLEAN_T send_check_writer(ErlNifEnv* env, + SocketDescriptor* descP, + ERL_NIF_TERM ref, + ERL_NIF_TERM* checkResult) +{ + if (descP->currentWriterP != NULL) { + ErlNifPid caller; + + if (enif_self(env, &caller) == NULL) { + *checkResult = esock_make_error(env, atom_exself); + return FALSE; + } + + if (!compare_pids(env, &descP->currentWriter.pid, &caller)) { + /* Not the "current writer", so (maybe) push onto queue */ + + SSDBG( descP, + ("SOCKET", "send_check_writer -> not (current) writer\r\n") ); + + if (!writer_search4pid(env, descP, &caller)) + *checkResult = writer_push(env, descP, caller, ref); + else + *checkResult = esock_make_error(env, esock_atom_eagain); + + SSDBG( descP, + ("SOCKET", + "nsend -> queue (push) result: %T\r\n", checkResult) ); + + return FALSE; + + } + + } + + *checkResult = esock_atom_ok; // Does not actually matter in this case, but ... + + return TRUE; +} + + + +/* *** send_check_result *** + * + * Check the result of a socket send (send, sendto and sendmsg) call. + * If a "complete" send has been made, the next (waiting) writer will be + * scheduled (if there is one). + * If we did not manage to send the entire package, make another select, + * so that we can be informed when we can make another try (to send the rest), + * and return with the amount we actually managed to send (its up to the caller + * (that is the erlang code) to figure out hust much is left to send). + * If the write fail, we give up and return with the appropriate error code. + * + * What about the remaining writers!! + */ static ERL_NIF_TERM send_check_result(ErlNifEnv* env, SocketDescriptor* descP, @@ -10437,24 +11005,67 @@ ERL_NIF_TERM send_check_result(ErlNifEnv* env, cnt_inc(&descP->writePkgCnt, 1); cnt_inc(&descP->writeByteCnt, written); + DEMONP(env, descP, &descP->currentWriter.mon); SSDBG( descP, ("SOCKET", "send_check_result -> " "everything written (%d,%d) - done\r\n", dataSize, written) ); + /* Ok, this write is done maybe activate the next (if any) */ + + if (writer_pop(env, descP, + &descP->currentWriter.pid, + &descP->currentWriter.mon, + &descP->currentWriter.ref)) { + + /* There was another one */ + + SSDBG( descP, ("SOCKET", "send_check_result -> new (active) writer: " + "\r\n pid: %T" + "\r\n ref: %T" + "\r\n", + descP->currentWriter.pid, + descP->currentWriter.ref) ); + + SELECT(env, + descP->sock, + (ERL_NIF_SELECT_WRITE), + descP, &descP->currentWriter.pid, descP->currentWriter.ref); + + } else { + descP->currentWriterP = NULL; + } + return esock_atom_ok; } else if (written < 0) { - /* Ouch, check what kind of failure */ + /* Some kind of send failure - check what kind */ + if ((saveErrno != EAGAIN) && (saveErrno != EINTR)) { + ErlNifPid pid; + ErlNifMonitor mon; + ERL_NIF_TERM ref, res; + + /* + * An actual failure - we (and everyone waiting) give up + */ cnt_inc(&descP->writeFails, 1); SSDBG( descP, ("SOCKET", "send_check_result -> error: %d\r\n", saveErrno) ); - return esock_make_error_errno(env, saveErrno); + res = esock_make_error_errno(env, saveErrno); + + while (writer_pop(env, descP, &pid, &mon, &ref)) { + SSDBG( descP, + ("SOCKET", "send_check_result -> abort %T\r\n", pid) ); + send_msg_nif_abort(env, ref, res, &pid); + DEMONP(env, descP, &mon); + } + + return res; } else { @@ -10531,8 +11142,9 @@ ERL_NIF_TERM recv_check_result(ErlNifEnv* env, } /* There is a special case: If the provided 'to read' value is - * zero (0). That means that we reads as much as we can, using - * the default read buffer size. + * zero (0) (only for type =/= stream). + * That means that we reads as much as we can, using the default + * read buffer size. */ if (bufP->size == read) { @@ -12541,6 +13153,8 @@ SocketDescriptor* alloc_descriptor(SOCKET sock, HANDLE event) if ((descP = enif_alloc_resource(sockets, sizeof(SocketDescriptor))) != NULL) { char buf[64]; /* Buffer used for building the mutex name */ + descP->env = enif_alloc_env(); + sprintf(buf, "socket[w,%d]", sock); descP->writeMtx = MCREATE(buf); descP->currentWriterP = NULL; // currentWriter not used @@ -13187,7 +13801,7 @@ ERL_NIF_TERM acceptor_push(ErlNifEnv* env, SocketRequestor* reqP = &e->data; reqP->pid = pid; - reqP->ref = ref; + reqP->ref = enif_make_copy(descP->env, ref); if (MONP(env, descP, &pid, &reqP->mon) > 0) { FREE(reqP); @@ -13231,6 +13845,108 @@ BOOLEAN_T acceptor_pop(ErlNifEnv* env, } +/* *** acceptor unqueue *** + * + * Remove an acceptor from the acceptor queue. + */ +static +BOOLEAN_T acceptor_unqueue(ErlNifEnv* env, + SocketDescriptor* descP, + const ErlNifPid* pid) +{ + return qunqueue(env, &descP->acceptorsQ, pid); +} + + + +/* *** writer search for pid *** + * + * Search for a pid in the writer queue. + */ +static +BOOLEAN_T writer_search4pid(ErlNifEnv* env, + SocketDescriptor* descP, + ErlNifPid* pid) +{ + return qsearch4pid(env, &descP->writersQ, pid); +} + + +/* *** writer push *** + * + * Push an writer onto the writer queue. + * This happens when we already have atleast one current writer. + */ +static +ERL_NIF_TERM writer_push(ErlNifEnv* env, + SocketDescriptor* descP, + ErlNifPid pid, + ERL_NIF_TERM ref) +{ + SocketRequestQueueElement* e = MALLOC(sizeof(SocketRequestQueueElement)); + SocketRequestor* reqP = &e->data; + + reqP->pid = pid; + reqP->ref = enif_make_copy(descP->env, ref); + + if (MONP(env, descP, &pid, &reqP->mon) > 0) { + FREE(reqP); + return esock_make_error(env, atom_exmon); + } + + qpush(&descP->writersQ, e); + + // THIS IS OK => MAKES THE CALLER WAIT FOR ITS TURN + return esock_make_error(env, esock_atom_eagain); +} + + +/* *** writer pop *** + * + * Pop an writer from the writer queue. + */ +static +BOOLEAN_T writer_pop(ErlNifEnv* env, + SocketDescriptor* descP, + ErlNifPid* pid, + ErlNifMonitor* mon, + ERL_NIF_TERM* ref) +{ + SocketRequestQueueElement* e = qpop(&descP->writersQ); + + if (e != NULL) { + *pid = e->data.pid; + *mon = e->data.mon; + *ref = e->data.ref; // At this point the ref has already been copied (env) + FREE(e); + return TRUE; + } else { + /* (acceptors) Queue was empty */ + // *pid = NULL; we have no null value for pids + // *mon = NULL; we have no null value for monitors + *ref = esock_atom_undefined; // Just in case + return FALSE; + } + +} + + +/* *** writer unqueue *** + * + * Remove an writer from the writer queue. + */ +static +BOOLEAN_T writer_unqueue(ErlNifEnv* env, + SocketDescriptor* descP, + const ErlNifPid* pid) +{ + return qunqueue(env, &descP->writersQ, pid); +} + + + + + static BOOLEAN_T qsearch4pid(ErlNifEnv* env, SocketRequestQueue* q, @@ -13673,7 +14389,7 @@ void socket_down(ErlNifEnv* env, "socket_down -> " "not current acceptor - maybe a waiting acceptor\r\n") ); - qunqueue(env, &descP->acceptorsQ, pid); + acceptor_unqueue(env, descP, pid); } } @@ -13723,7 +14439,7 @@ ErlNifFunc socket_funcs[] = * is called after the connect *select* has "completed". */ {"nif_finalize_connection", 1, nif_finalize_connection, 0}, - // {"nif_cancel", 2, nif_cancel, 0}, + {"nif_cancel", 3, nif_cancel, 0}, {"nif_finalize_close", 1, nif_finalize_close, ERL_NIF_DIRTY_JOB_IO_BOUND} }; @@ -13843,8 +14559,10 @@ int on_load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info) atom_want = MKA(env, str_want); /* Global atom(s) */ + esock_atom_accept = MKA(env, "accept"); esock_atom_addr = MKA(env, "addr"); esock_atom_any = MKA(env, "any"); + esock_atom_connect = MKA(env, "connect"); esock_atom_credentials = MKA(env, "credentials"); esock_atom_ctrl = MKA(env, "ctrl"); esock_atom_ctrunc = MKA(env, "ctrunc"); @@ -13869,6 +14587,7 @@ int on_load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info) esock_atom_loopback = MKA(env, "loopback"); esock_atom_lowdelay = MKA(env, "lowdelay"); esock_atom_mincost = MKA(env, "mincost"); + esock_atom_not_found = MKA(env, "not_found"); esock_atom_ok = MKA(env, "ok"); esock_atom_oob = MKA(env, "oob"); esock_atom_origdstaddr = MKA(env, "origdstaddr"); @@ -13878,11 +14597,18 @@ int on_load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info) esock_atom_protocol = MKA(env, "protocol"); esock_atom_raw = MKA(env, "raw"); esock_atom_rdm = MKA(env, "rdm"); + esock_atom_recv = MKA(env, "recv"); + esock_atom_recvfrom = MKA(env, "recvfrom"); + esock_atom_recvmsg = MKA(env, "recvmsg"); esock_atom_reliability = MKA(env, "reliability"); esock_atom_rights = MKA(env, "rights"); esock_atom_scope_id = MKA(env, "scope_id"); esock_atom_sctp = MKA(env, "sctp"); esock_atom_sec = MKA(env, "sec"); + esock_atom_select_sent = MKA(env, "select_sent"); + esock_atom_send = MKA(env, "send"); + esock_atom_sendmsg = MKA(env, "sendmsg"); + esock_atom_sendto = MKA(env, "sendto"); esock_atom_seqpacket = MKA(env, "seqpacket"); esock_atom_socket = MKA(env, "socket"); esock_atom_spec_dst = MKA(env, "spec_dst"); |