X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Ftransport%2Fplugin_transport_http.c;h=f71acf4b0cbb0c703748bc67a60a900962747a9c;hb=aaf312180fd09f08e30316a0a646a662efb29219;hp=78e513857db3cfeb888505858b8614137619530a;hpb=5f1e438c5995d178d25f21f199a0026eaf631dd8;p=oweals%2Fgnunet.git diff --git a/src/transport/plugin_transport_http.c b/src/transport/plugin_transport_http.c index 78e513857..f71acf4b0 100644 --- a/src/transport/plugin_transport_http.c +++ b/src/transport/plugin_transport_http.c @@ -40,8 +40,9 @@ #include -#define DEBUG_CURL GNUNET_YES +#define DEBUG_CURL GNUNET_NO #define DEBUG_HTTP GNUNET_NO +#define DEBUG_CONNECTIONS GNUNET_YES #define INBOUND GNUNET_NO #define OUTBOUND GNUNET_YES @@ -135,7 +136,7 @@ struct HTTP_Message * buffer length */ size_t size; - + /** * Continuation function to call once the transmission buffer * has again space available. NULL if there is no @@ -152,91 +153,61 @@ struct HTTP_Message struct HTTP_PeerContext { - struct GNUNET_PeerIdentity identity; - - struct HTTP_Session * head; - struct HTTP_Session * tail; -}; - - -struct HTTP_Session -{ - struct HTTP_Session * next; - struct HTTP_Session * prev; - - void * addr; - size_t addrlen; - /** - * target url + * peer's identity */ - char * url; + struct GNUNET_PeerIdentity identity; /** - * Message queue for outbound messages - * head of queue + * Pointer to the global plugin struct. */ - struct HTTP_Message * pending_msgs_head; + struct Plugin *plugin; /** - * Message queue for outbound messages - * tail of queue + * Linked list of connections with this peer + * head */ - struct HTTP_Message * pending_msgs_tail; + struct Session * head; /** - * partner peer this connection belongs to + * Linked list of connections with this peer + * tail */ - struct HTTP_PeerContext * peercontext; + struct Session * tail; /** - * message stream tokenizer for incoming data + * id for next session */ - struct GNUNET_SERVER_MessageStreamTokenizer *msgtok; + size_t session_id_counter; +}; + +struct Session +{ /** - * session direction - * outbound: GNUNET_YES - * inbound : GNUNET_NO + * API requirement. */ - unsigned int direction; - - unsigned int send_connected; - unsigned int send_active; - unsigned int recv_connected; - unsigned int recv_active; + struct SessionHeader header; /** - * entity managing sending data + * next session in linked list */ - void * send_endpoint; + struct Session * next; /** - * entity managing recieving data + * previous session in linked list */ - void * recv_endpoint; -}; - - -struct HTTP_Connection -{ - struct HTTP_Connection * next; - struct HTTP_Connection * prev; - - void * addr; - size_t addrlen; + struct Session * prev; /** - * Message queue for outbound messages - * head of queue + * address of this session */ - struct HTTP_Message * pending_msgs_head; + void * addr; /** - * Message queue for outbound messages - * tail of queue + * address length */ - struct HTTP_Message * pending_msgs_tail; + size_t addrlen; /** * target url @@ -244,122 +215,83 @@ struct HTTP_Connection char * url; /** - * status of PUT connection - * connected? - */ - unsigned int put_connected; - - /** - * status of PUT connection - * if no data to send, connection is paused + * Message queue for outbound messages + * head of queue */ - unsigned int put_send_paused; - - unsigned int get_connected; + struct HTTP_Message * pending_msgs_head; /** - * curl handle for sending data using HTTP/PUT - * outbound data + * Message queue for outbound messages + * tail of queue */ - CURL *put_curl_handle; + struct HTTP_Message * pending_msgs_tail; /** - * curl handle for recieving data using HTTP/GET transmission - * inbound data + * partner peer this connection belongs to */ - CURL *get_curl_handle; - - struct Session * session; - - struct GNUNET_SERVER_MessageStreamTokenizer * get_msgtok; -}; - -struct HTTP_Connection_in -{ - struct HTTP_Connection_in * next; - - struct HTTP_Connection_in * prev; - - void * addr; - size_t addrlen; - - unsigned int connected; - unsigned int send_paused; - - struct GNUNET_SERVER_MessageStreamTokenizer * msgtok; - - struct Session * session; + struct HTTP_PeerContext * peercontext; /** - * Is there a HTTP/PUT in progress? + * message stream tokenizer for incoming data */ - int is_put_in_progress; + struct GNUNET_SERVER_MessageStreamTokenizer *msgtok; /** - * Is the http request invalid? + * session direction + * outbound: OUTBOUND (GNUNET_YES) + * inbound : INBOUND (GNUNET_NO) */ - int is_bad_request; -}; - -/** - * Session handle for connections. - */ -struct Session -{ + unsigned int direction; /** - * API requirement. + * is session connected to send data? */ - struct SessionHeader header; + unsigned int send_connected; /** - * Stored in a linked list. + * is send connection active? */ - struct Session *next; + unsigned int send_active; /** - * Pointer to the global plugin struct. + * connection disconnect forced (e.g. from transport) */ - struct Plugin *plugin; + unsigned int send_force_disconnect; /** - * To whom are we talking to (set to our identity - * if we are still waiting for the welcome message) + * is session connected to receive data? */ - struct GNUNET_PeerIdentity identity; + unsigned int recv_connected; /** - * Did we initiate the connection (GNUNET_YES) or the other peer (GNUNET_NO)? + * is receive connection active? */ - int is_client; + unsigned int recv_active; /** - * At what time did we reset last_received last? + * connection disconnect forced (e.g. from transport) */ - struct GNUNET_TIME_Absolute last_quota_update; + unsigned int recv_force_disconnect; /** - * How many bytes have we received since the "last_quota_update" - * timestamp? + * id for next session + * NOTE: 0 is not an ID, zero is not defined. A correct ID is always > 0 */ - uint64_t last_received; + size_t session_id; /** - * Number of bytes per ms that this peer is allowed - * to send to us. + * entity managing sending data + * outbound session: CURL * + * inbound session: mhd_connection * */ - uint32_t quota; + void * send_endpoint; /** - * Encoded hash + * entity managing recieving data + * outbound session: CURL * + * inbound session: mhd_connection * */ - struct GNUNET_CRYPTO_HashAsciiEncoded hash; - - struct HTTP_Connection *outbound_connections_head; - struct HTTP_Connection *outbound_connections_tail; - - struct HTTP_Connection_in *inbound_connections_head; - struct HTTP_Connection_in *inbound_connections_tail; + void * recv_endpoint; }; /** @@ -374,11 +306,6 @@ struct Plugin unsigned int port_inbound; - /** - * Hashmap for all existing sessions. - */ - struct GNUNET_CONTAINER_MultiHashMap *sessions; - struct GNUNET_CONTAINER_MultiHashMap *peers; /** @@ -435,73 +362,16 @@ http_plugin_address_to_string (void *cls, const void *addr, size_t addrlen); -/** - * Create a new session - * @param cls plugin as closure - * @param addr_in address the peer is using inbound - * @param addr_len_in address length - * @param addr_out address the peer is using outbound - * @param addr_len_out address length - * @param peer identity - * @return created session object - */ -static struct Session * -create_session (void * cls, - char * addr_in, - size_t addrlen_in, - char * addr_out, - size_t addrlen_out, - const struct GNUNET_PeerIdentity *peer) -{ - struct Plugin *plugin = cls; - struct Session * cs = GNUNET_malloc ( sizeof( struct Session) ); - - GNUNET_assert(cls !=NULL); - cs->plugin = plugin; - memcpy(&cs->identity, peer, sizeof (struct GNUNET_PeerIdentity)); - GNUNET_CRYPTO_hash_to_enc(&cs->identity.hashPubKey,&(cs->hash)); - cs->outbound_connections_head = NULL; - cs->outbound_connections_tail = NULL; - return cs; -} - -/** - * Check if session for this peer is already existing, otherwise create it - * @param cls the plugin used - * @param p peer to get session for - * @return session found or created - */ -static struct Session * session_get (void * cls, const struct GNUNET_PeerIdentity *p) -{ - struct Plugin *plugin = cls; - struct Session *cs; - unsigned int res; - - cs = GNUNET_CONTAINER_multihashmap_get (plugin->sessions, &p->hashPubKey); - if (cs == NULL) - { - cs = create_session(plugin, NULL, 0, NULL, 0, p); - res = GNUNET_CONTAINER_multihashmap_put ( plugin->sessions, - &cs->identity.hashPubKey, - cs, - GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY); - if (res == GNUNET_OK) - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "New Session `%s' inserted\n", GNUNET_i2s(p)); - } - return cs; -} - -static char * create_url(void * cls, const void * addr, size_t addrlen) +static char * create_url(void * cls, const void * addr, size_t addrlen, size_t id) { struct Plugin *plugin = cls; char *url = NULL; GNUNET_assert ((addr!=NULL) && (addrlen != 0)); GNUNET_asprintf(&url, - "http://%s/%s", + "http://%s/%s;%u", http_plugin_address_to_string(NULL, addr, addrlen), - (char *) (&plugin->my_ascii_hash_ident)); + (char *) (&plugin->my_ascii_hash_ident),id); return url; } @@ -512,8 +382,7 @@ static char * create_url(void * cls, const void * addr, size_t addrlen) * @param msg message to remove * @return GNUNET_SYSERR if msg not found, GNUNET_OK on success */ - -static int remove_http_message(struct HTTP_Session * ps, struct HTTP_Message * msg) +static int remove_http_message (struct Session * ps, struct HTTP_Message * msg) { GNUNET_CONTAINER_DLL_remove(ps->pending_msgs_head,ps->pending_msgs_tail,msg); GNUNET_free(msg); @@ -521,57 +390,42 @@ static int remove_http_message(struct HTTP_Session * ps, struct HTTP_Message * m } /** - * Check if session already knows this address for a outbound connection to this peer - * If address not in session, add it to the session - * @param cls the plugin used - * @param cs the session - * @param addr address - * @param addr_len address length - * @return the found or created address + * Removes a session from the linked list of sessions + * @param pc peer context + * @param ps session + * @param call_msg_cont GNUNET_YES to call pending message continuations, otherwise no + * @param call_msg_cont_result, result to call message continuations with + * @return GNUNET_SYSERR if msg not found, GNUNET_OK on success */ -static struct HTTP_Connection * session_check_outbound_address (void * cls, struct Session *cs, const void * addr, size_t addr_len) +static int remove_session (struct HTTP_PeerContext * pc, struct Session * ps, int call_msg_cont, int call_msg_cont_result) { - struct Plugin *plugin = cls; - struct HTTP_Connection * cc = cs->outbound_connections_head; - struct HTTP_Connection * con = NULL; - - GNUNET_assert((addr_len == sizeof (struct IPv4HttpAddress)) || (addr_len == sizeof (struct IPv6HttpAddress))); + struct HTTP_Message * msg; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: removing %s session with id %u\n", ps, (ps->direction == INBOUND) ? "inbound" : "outbound",ps->session_id); + GNUNET_free_non_null (ps->addr); + GNUNET_SERVER_mst_destroy (ps->msgtok); + GNUNET_free(ps->url); - while (cc!=NULL) + msg = ps->pending_msgs_head; + while (msg!=NULL) { - if (addr_len == cc->addrlen) + if ((call_msg_cont == GNUNET_YES) && (msg->transmit_cont!=NULL)) { - if (0 == memcmp(cc->addr, addr, addr_len)) - { - con = cc; - break; - } + msg->transmit_cont (msg->transmit_cont_cls,&pc->identity,call_msg_cont_result); } - cc=cc->next; + GNUNET_free(msg); + GNUNET_CONTAINER_DLL_remove(ps->pending_msgs_head,ps->pending_msgs_head,msg); } - if (con==NULL) - { - con = GNUNET_malloc(sizeof(struct HTTP_Connection) + addr_len); - con->addrlen = addr_len; - con->addr=&con[1]; - con->url=create_url(plugin, addr, addr_len); - con->put_curl_handle = NULL; - con->put_connected = GNUNET_NO; - con->get_curl_handle = NULL; - con->get_connected = GNUNET_NO; - con->session = cs; - memcpy(con->addr, addr, addr_len); - GNUNET_CONTAINER_DLL_insert(cs->outbound_connections_head,cs->outbound_connections_tail,con); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Created new connection %X to peer `%s'\n",con,GNUNET_i2s(&cs->identity)); - } - return con; + GNUNET_CONTAINER_DLL_remove(pc->head,pc->tail,ps); + GNUNET_free(ps); + ps = NULL; + return GNUNET_OK; } -static struct HTTP_Session * get_HTTP_Session (void * cls, struct HTTP_PeerContext *pc, const void * addr, size_t addr_len) +static struct Session * get_Session (void * cls, struct HTTP_PeerContext *pc, const void * addr, size_t addr_len) { - struct HTTP_Session * cc = pc->head; - struct HTTP_Session * con = NULL; + struct Session * cc = pc->head; + struct Session * con = NULL; unsigned int count = 0; GNUNET_assert((addr_len == sizeof (struct IPv4HttpAddress)) || (addr_len == sizeof (struct IPv6HttpAddress))); @@ -581,7 +435,9 @@ static struct HTTP_Session * get_HTTP_Session (void * cls, struct HTTP_PeerConte { if (0 == memcmp(cc->addr, addr, addr_len)) { - con = cc; + /* connection can not be used, since it is disconnected */ + if ((cc->recv_force_disconnect==GNUNET_NO) && (cc->send_force_disconnect==GNUNET_NO)) + con = cc; break; } } @@ -593,108 +449,86 @@ static struct HTTP_Session * get_HTTP_Session (void * cls, struct HTTP_PeerConte /** - * Check if session already knows this address for a inbound connection to this peer - * If address not in session, add it to the session - * @param cls not needed, can be NULL - * @param cs the session - * @param addr address - * @param addr_len address length - * @return the found or created address + * Callback called by MHD when a connection is terminated */ -static struct HTTP_Connection_in * session_check_inbound_address (void * cls, struct Session *cs, const void * addr, size_t addr_len) +static void mhd_termination_cb (void *cls, struct MHD_Connection * connection, void **httpSessionCache) { - struct HTTP_Connection_in * cc = cs->inbound_connections_head; - struct HTTP_Connection_in * con = NULL; - - GNUNET_assert((addr_len == sizeof (struct IPv4HttpAddress)) || (addr_len == sizeof (struct IPv6HttpAddress))); + struct Session * ps = *httpSessionCache; + if (ps == NULL) + return; + struct HTTP_PeerContext * pc = ps->peercontext; - while (cc!=NULL) + if (connection==ps->recv_endpoint) { - if (addr_len == cc->addrlen) - { - if (0 == memcmp(cc->addr, addr, addr_len)) - { - con = cc; - break; - } - } - cc=cc->next; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: inbound connection from peer `%s' was terminated\n", ps, GNUNET_i2s(&pc->identity)); +#endif + ps->recv_active = GNUNET_NO; + ps->recv_connected = GNUNET_NO; + ps->recv_endpoint = NULL; } + if (connection==ps->send_endpoint) + { + ps->send_active = GNUNET_NO; + ps->send_connected = GNUNET_NO; + ps->send_endpoint = NULL; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: outbound connection from peer `%s' was terminated\n", ps, GNUNET_i2s(&pc->identity)); +#endif + } - if (con==NULL) + /* if both connections disconnected, remove session */ + if ((ps->send_connected == GNUNET_NO) && (ps->recv_connected == GNUNET_NO)) { - con = GNUNET_malloc(sizeof(struct HTTP_Connection_in) + addr_len); - con->addrlen = addr_len; - con->addr=&con[1]; - con->connected = GNUNET_NO; - con->session = cs; - memcpy(con->addr, addr, addr_len); - GNUNET_CONTAINER_DLL_insert(cs->inbound_connections_head,cs->inbound_connections_tail,con); + remove_session(pc,ps,GNUNET_YES,GNUNET_SYSERR); } - - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X for inbound address %s (%s) was found\n", - con, - GNUNET_i2s(&cs->identity), - http_plugin_address_to_string(NULL,con->addr,con->addrlen)); - return con; -} - - -/** - * Callback called by MHD when a connection is terminated - */ -static void requestCompletedCallback (void *cls, struct MHD_Connection * connection, void **httpSessionCache) -{ - struct HTTP_Connection_in * con; - - con = *httpSessionCache; - if (con == NULL) - return; - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection from peer `%s' was terminated\n",GNUNET_i2s(&con->session->identity)); - /* session set to inactive */ - con->is_put_in_progress = GNUNET_NO; - con->is_bad_request = GNUNET_NO; } static void mhd_write_mst_cb (void *cls, void *client, const struct GNUNET_MessageHeader *message) { - struct HTTP_Connection_in * con = cls; - GNUNET_assert(con != NULL); + + struct Session *ps = cls; + struct HTTP_PeerContext *pc = ps->peercontext; + GNUNET_assert(ps != NULL); + GNUNET_assert(pc != NULL); GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Forwarding message to transport service, type %u and size %u from `%s' (`%s')\n", + "Connection %X: Forwarding message to transport service, type %u and size %u from `%s' (`%s')\n", + ps, ntohs(message->type), ntohs(message->size), - GNUNET_i2s(&(con->session->identity)),http_plugin_address_to_string(NULL,con->addr,con->addrlen)); + GNUNET_i2s(&(ps->peercontext)->identity),http_plugin_address_to_string(NULL,ps->addr,ps->addrlen)); - con->session->plugin->env->receive (con->session->plugin->env->cls, - &con->session->identity, - message, 1, con->session, - NULL, - 0); + pc->plugin->env->receive (ps->peercontext->plugin->env->cls, + &pc->identity, + message, 1, ps, + ps->addr, + ps->addrlen); } -static void curl_write_mst_cb (void *cls, +static void curl_receive_mst_cb (void *cls, void *client, const struct GNUNET_MessageHeader *message) { - struct HTTP_Connection * con = cls; - GNUNET_assert(con != NULL); + struct Session *ps = cls; + struct HTTP_PeerContext *pc = ps->peercontext; + GNUNET_assert(ps != NULL); + GNUNET_assert(pc != NULL); GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Forwarding message to transport service, type %u and size %u from `%s' (`%s')\n", ntohs(message->type), ntohs(message->size), - GNUNET_i2s(&(con->session->identity)),http_plugin_address_to_string(NULL,con->addr,con->addrlen)); + GNUNET_i2s(&(pc->identity)),http_plugin_address_to_string(NULL,ps->addr,ps->addrlen)); - con->session->plugin->env->receive (con->session->plugin->env->cls, - &con->session->identity, - message, 1, con->session, - con->addr, - con->addrlen); + pc->plugin->env->receive (pc->plugin->env->cls, + &pc->identity, + message, 1, ps, + ps->addr, + ps->addrlen); } @@ -702,7 +536,7 @@ static void curl_write_mst_cb (void *cls, * Check if ip is allowed to connect. */ static int -acceptPolicyCallback (void *cls, +mhd_accept_cb (void *cls, const struct sockaddr *addr, socklen_t addr_len) { #if 0 @@ -712,39 +546,53 @@ acceptPolicyCallback (void *cls, return MHD_YES; } -int server_read_callback (void *cls, uint64_t pos, char *buf, int max) +int mhd_send_callback (void *cls, uint64_t pos, char *buf, int max) { int bytes_read = 0; - struct HTTP_Connection * con = cls; + struct Session * ps = cls; + struct HTTP_PeerContext * pc; struct HTTP_Message * msg; int res;res=5; - msg=con->pending_msgs_tail; + GNUNET_assert (ps!=NULL); + pc = ps->peercontext; + msg = ps->pending_msgs_tail; + if (ps->send_force_disconnect==GNUNET_YES) + { +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: outbound forced to disconnect\n",ps); +#endif + return -1; + } if (msg!=NULL) { - /* + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"mhd_send_callback %X: queue msg size: %u, max %u pos %u msg->pos %u\n",ps,msg->size,max,pos,msg->pos); if ((msg->size-msg->pos) <= max) - { - memcpy(buf,&msg->buf[pos],(msg->size-msg->pos)); - pos+=(msg->size-msg->pos); - } - else - { - memcpy(buf,&msg->buf[pos],max); - pos+=max; - } + { + memcpy(buf,&msg->buf[msg->pos],(msg->size-msg->pos)); + bytes_read = msg->size-msg->pos; + msg->pos+=(msg->size-msg->pos); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"mhd_send_callback %X: complete: size %u pos %u bytes read %u \n",ps,msg->size,msg->pos,bytes_read); + } + else + { + memcpy(buf,&msg->buf[msg->pos],max); + msg->pos+=max; + bytes_read = max; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"mhd_send_callback %X: partial: size %u pos %u bytes read %u \n",ps,msg->size,msg->pos,bytes_read); + } - if (msg->pos==msg->size) - { - if (NULL != con->pending_msgs_tail->transmit_cont) - msg->transmit_cont (msg->transmit_cont_cls,&con->session->identity,GNUNET_SYSERR); - res = remove_http_message(con,msg); - } - */ + if (msg->pos==msg->size) + { + struct GNUNET_MessageHeader * tmp = (struct GNUNET_MessageHeader *) msg->buf; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"MHD SENT MESSAGE %u bytes msg->type %u msg->size %u\n", bytes_read, ntohs(tmp->type), ntohs(tmp->size)); + if (NULL!=msg->transmit_cont) + msg->transmit_cont (msg->transmit_cont_cls,&pc->identity,GNUNET_OK); + res = remove_http_message(ps,msg); + } } - return bytes_read; } @@ -756,7 +604,7 @@ int server_read_callback (void *cls, uint64_t pos, char *buf, int max) * already exists and create a new one if not. */ static int -accessHandlerCallback (void *cls, +mdh_access_cb (void *cls, struct MHD_Connection *mhd_connection, const char *url, const char *method, @@ -766,28 +614,47 @@ accessHandlerCallback (void *cls, { struct Plugin *plugin = cls; struct MHD_Response *response; - struct Session * cs; - struct HTTP_Connection_in * con; const union MHD_ConnectionInfo * conn_info; + struct sockaddr_in *addrin; struct sockaddr_in6 *addrin6; + char address[INET6_ADDRSTRLEN+14]; struct GNUNET_PeerIdentity pi_in; - int res = GNUNET_NO; - int send_error_to_client; + size_t id_num = 0; + struct IPv4HttpAddress ipv4addr; struct IPv6HttpAddress ipv6addr; + struct HTTP_PeerContext *pc; - struct HTTP_Session *ps; + struct Session *ps; + struct Session *ps_tmp; + + int res = GNUNET_NO; + int send_error_to_client; void * addr; size_t addr_len; GNUNET_assert(cls !=NULL); send_error_to_client = GNUNET_NO; + if (NULL == *httpSessionCache) { /* check url for peer identity , if invalid send HTTP 404*/ - res = GNUNET_CRYPTO_hash_from_string ( &url[1], &(pi_in.hashPubKey)); + size_t len = strlen(&url[1]); + char * peer = GNUNET_malloc(104+1); + + if ((len>104) && (url[104]==';')) + { + char * id = GNUNET_malloc((len-104)+1); + strcpy(id,&url[105]); + memcpy(peer,&url[1],103); + peer[103] = '\0'; + id_num = strtoul ( id, NULL , 10); + GNUNET_free(id); + } + res = GNUNET_CRYPTO_hash_from_string (peer, &(pi_in.hashPubKey)); + GNUNET_free(peer); if ( GNUNET_SYSERR == res ) { response = MHD_create_response_from_data (strlen (HTTP_ERROR_RESPONSE),HTTP_ERROR_RESPONSE, MHD_NO, MHD_NO); @@ -802,29 +669,23 @@ accessHandlerCallback (void *cls, } else { - con = *httpSessionCache; - cs = con->session; + ps = *httpSessionCache; + pc = ps->peercontext; } if (NULL == *httpSessionCache) { - /* get session for peer identity */ - cs = session_get (plugin ,&pi_in); - /* get peer context */ pc = GNUNET_CONTAINER_multihashmap_get (plugin->peers, &pi_in.hashPubKey); /* Peer unknown */ if (pc==NULL) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"RECV: CREATING NEW PEER CONTEXT\n"); pc = GNUNET_malloc(sizeof (struct HTTP_PeerContext)); + pc->plugin = plugin; + pc->session_id_counter=1; memcpy(&pc->identity, &pi_in, sizeof(struct GNUNET_PeerIdentity)); GNUNET_CONTAINER_multihashmap_put(plugin->peers, &pc->identity.hashPubKey, pc, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY); } - else - { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"RECV: PEER CONTEXT FOUND\n"); - } conn_info = MHD_get_connection_info(mhd_connection, MHD_CONNECTION_INFO_CLIENT_ADDRESS ); /* Incoming IPv4 connection */ @@ -836,7 +697,6 @@ accessHandlerCallback (void *cls, ipv4addr.u_port = addrin->sin_port; addr = &ipv4addr; addr_len = sizeof(struct IPv4HttpAddress); - con = session_check_inbound_address (plugin, cs, (const void *) &ipv4addr, sizeof (struct IPv4HttpAddress)); } /* Incoming IPv6 connection */ if ( AF_INET6 == conn_info->client_addr->sin_family) @@ -847,67 +707,94 @@ accessHandlerCallback (void *cls, ipv6addr.u6_port = addrin6->sin6_port; addr = &ipv6addr; addr_len = sizeof(struct IPv6HttpAddress); - con = session_check_inbound_address (plugin, cs, &ipv6addr, sizeof (struct IPv6HttpAddress)); } - /* Set closure and update current session*/ - ps = get_HTTP_Session(plugin, pc, addr, addr_len); + //ps = get_Session(plugin, pc, addr, addr_len); + ps = NULL; + /* only inbound sessions here */ + + ps_tmp = pc->head; + while (ps_tmp!=NULL) + { + if ((ps_tmp->direction==INBOUND) && (ps_tmp->session_id == id_num) && (id_num!=0)) + { + if ((ps_tmp->recv_force_disconnect!=GNUNET_YES) && (ps_tmp->send_force_disconnect!=GNUNET_YES)) + ps=ps_tmp; + break; + } + ps_tmp=ps_tmp->next; + } + if (ps==NULL) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"RECV: CREATING NEW SESSION %s\n",http_plugin_address_to_string(NULL, addr, addr_len)); - ps = GNUNET_malloc(sizeof (struct HTTP_Session)); + ps = GNUNET_malloc(sizeof (struct Session)); ps->addr = GNUNET_malloc(addr_len); memcpy(ps->addr,addr,addr_len); ps->addrlen = addr_len; - ps->direction=GNUNET_NO; + ps->direction=INBOUND; ps->pending_msgs_head = NULL; ps->pending_msgs_tail = NULL; - ps->url = create_url (plugin, ps->addr, ps->addrlen); + ps->send_connected=GNUNET_NO; + ps->send_active=GNUNET_NO; + ps->recv_connected=GNUNET_NO; + ps->recv_active=GNUNET_NO; + ps->peercontext=pc; + ps->session_id =id_num; + ps->url = create_url (plugin, ps->addr, ps->addrlen, ps->session_id); GNUNET_CONTAINER_DLL_insert(pc->head,pc->tail,ps); } - else - { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"RECV: SESSION CONTEXT FOUND\n"); - } - *httpSessionCache = con; - if (con->msgtok==NULL) - con->msgtok = GNUNET_SERVER_mst_create (&mhd_write_mst_cb, con); + *httpSessionCache = ps; + if (ps->msgtok==NULL) + ps->msgtok = GNUNET_SERVER_mst_create (&mhd_write_mst_cb, ps); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"HTTP Daemon has new an incoming `%s' request from peer `%s' (`%s')\n", + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: HTTP Daemon has new an incoming `%s' request from peer `%s' (`%s')\n", + ps, method, - GNUNET_i2s(&cs->identity), - http_plugin_address_to_string(NULL, con->addr, con->addrlen)); + GNUNET_i2s(&pc->identity), + http_plugin_address_to_string(NULL, ps->addr, ps->addrlen)); } /* Is it a PUT or a GET request */ if (0 == strcmp (MHD_HTTP_METHOD_PUT, method)) { - if ((*upload_data_size == 0) && (con->is_put_in_progress==GNUNET_NO)) + if (ps->recv_force_disconnect) + { +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: inbound connection was forced to disconnect\n",ps); +#endif + ps->recv_active = GNUNET_NO; + return MHD_NO; + } + if ((*upload_data_size == 0) && (ps->recv_active==GNUNET_NO)) { - con->is_put_in_progress = GNUNET_YES; + ps->recv_endpoint = mhd_connection; + ps->recv_connected = GNUNET_YES; + ps->recv_active = GNUNET_YES; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: inbound PUT connection connected\n",ps); +#endif return MHD_YES; } /* Transmission of all data complete */ - if ((*upload_data_size == 0) && (con->is_put_in_progress == GNUNET_YES)) + if ((*upload_data_size == 0) && (ps->recv_active == GNUNET_YES)) { - response = MHD_create_response_from_data (strlen (HTTP_PUT_RESPONSE),HTTP_PUT_RESPONSE, MHD_NO, MHD_NO); - res = MHD_queue_response (mhd_connection, MHD_HTTP_OK, response); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Sent HTTP/1.1: 200 OK as PUT Response\n",HTTP_PUT_RESPONSE, strlen (HTTP_PUT_RESPONSE), res ); - MHD_destroy_response (response); - return MHD_YES; - - con->is_put_in_progress = GNUNET_NO; - con->is_bad_request = GNUNET_NO; - return res; + response = MHD_create_response_from_data (strlen (HTTP_PUT_RESPONSE),HTTP_PUT_RESPONSE, MHD_NO, MHD_NO); + res = MHD_queue_response (mhd_connection, MHD_HTTP_OK, response); +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: Sent HTTP/1.1: 200 OK as PUT Response\n",ps); +#endif + MHD_destroy_response (response); + ps->recv_active=GNUNET_NO; + return MHD_YES; } /* Recieving data */ - if ((*upload_data_size > 0) && (con->is_put_in_progress == GNUNET_YES)) + if ((*upload_data_size > 0) && (ps->recv_active == GNUNET_YES)) { - res = GNUNET_SERVER_mst_receive(con->msgtok, con, upload_data,*upload_data_size, GNUNET_NO, GNUNET_NO); + res = GNUNET_SERVER_mst_receive(ps->msgtok, ps, upload_data,*upload_data_size, GNUNET_NO, GNUNET_NO); (*upload_data_size) = 0; return MHD_YES; } @@ -916,18 +803,24 @@ accessHandlerCallback (void *cls, } if ( 0 == strcmp (MHD_HTTP_METHOD_GET, method) ) { - response = MHD_create_response_from_callback(-1,32 * 1024, &server_read_callback, con, NULL); + if (ps->send_force_disconnect) + { +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: outbound connection was forced to disconnect\n",ps); +#endif + ps->send_active = GNUNET_NO; + return MHD_NO; + } + ps->send_connected = GNUNET_YES; + ps->send_active = GNUNET_YES; + ps->send_endpoint = mhd_connection; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: inbound GET connection connected\n",ps); +#endif + response = MHD_create_response_from_callback(-1,32 * 1024, &mhd_send_callback, ps, NULL); res = MHD_queue_response (mhd_connection, MHD_HTTP_OK, response); MHD_destroy_response (response); - - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"HTTP Daemon has new an incoming `%s' request from peer `%s' (`%s')\n", - method, - GNUNET_i2s(&cs->identity), - http_plugin_address_to_string(NULL, con->addr, con->addrlen)); - - - return res; - + return MHD_YES; } return MHD_NO; } @@ -1058,16 +951,23 @@ static void http_server_daemon_v6_run (void *cls, return; } +/** + * Function setting up curl handle and selecting message to send + * @param cls plugin + * @param ses session to send data to + * @param con connection + * @return bytes sent to peer + */ +static ssize_t send_check_connections (void *cls, struct Session *ps); -static size_t curl_header_function( void *ptr, size_t size, size_t nmemb, void *stream) +static size_t curl_get_header_function( void *ptr, size_t size, size_t nmemb, void *stream) { - struct HTTP_Session * ps = stream; + struct Session * ps = stream; char * tmp; size_t len = size * nmemb; long http_result = 0; int res; - /* Getting last http result code */ if (ps->recv_connected==GNUNET_NO) { @@ -1078,7 +978,12 @@ static size_t curl_header_function( void *ptr, size_t size, size_t nmemb, void * if (http_result == 200) { ps->recv_connected = GNUNET_YES; - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: inbound connected\n",ps); + ps->recv_active = GNUNET_YES; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: connected to recieve data\n",ps); +#endif + // Calling send_check_connections again since receive is established + send_check_connections (ps->peercontext->plugin, ps); } } } @@ -1105,6 +1010,60 @@ static size_t curl_header_function( void *ptr, size_t size, size_t nmemb, void * return size * nmemb; } +static size_t curl_put_header_function( void *ptr, size_t size, size_t nmemb, void *stream) +{ + struct Session * ps = stream; + + char * tmp; + size_t len = size * nmemb; + long http_result = 0; + int res; + + /* Getting last http result code */ + GNUNET_assert(NULL!=ps); + res = curl_easy_getinfo(ps->send_endpoint, CURLINFO_RESPONSE_CODE, &http_result); + if (CURLE_OK == res) + { + if ((http_result == 100) && (ps->send_connected==GNUNET_NO)) + { + ps->send_connected = GNUNET_YES; + ps->send_active = GNUNET_YES; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: connected to send data\n",ps); +#endif + } + if ((http_result == 200) && (ps->send_connected==GNUNET_YES)) + { + ps->send_connected = GNUNET_NO; + ps->send_active = GNUNET_NO; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: sending disconnected\n",ps); +#endif + } + } + + tmp = NULL; + if ((size * nmemb) < SIZE_MAX) + tmp = GNUNET_malloc (len+1); + + if ((tmp != NULL) && (len > 0)) + { + memcpy(tmp,ptr,len); + if (len>=2) + { + if (tmp[len-2] == 13) + tmp[len-2]= '\0'; + } +#if DEBUG_HTTP + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Header: `%s' %u \n",tmp, http_result); +#endif + } + if (NULL != tmp) + GNUNET_free (tmp); + + return size * nmemb; +} + /** * Callback method used with libcurl * Method is called when libcurl needs to read data during sending @@ -1114,16 +1073,18 @@ static size_t curl_header_function( void *ptr, size_t size, size_t nmemb, void * * @param ptr source pointer, passed to the libcurl handle * @return bytes written to stream */ -static size_t send_curl_read_callback(void *stream, size_t size, size_t nmemb, void *ptr) +static size_t curl_send_cb(void *stream, size_t size, size_t nmemb, void *ptr) { - struct HTTP_Session * ps = ptr; + struct Session * ps = ptr; struct HTTP_Message * msg = ps->pending_msgs_tail; size_t bytes_sent; size_t len; if (ps->pending_msgs_tail == NULL) { +#if DEBUG_CONNECTIONS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: No Message to send, pausing connection\n",ps); +#endif ps->send_active = GNUNET_NO; return CURL_READFUNC_PAUSE; } @@ -1156,7 +1117,9 @@ static size_t send_curl_read_callback(void *stream, size_t size, size_t nmemb, v if ( msg->pos == msg->size) { +#if DEBUG_CONNECTIONS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: Message with %u bytes sent, removing message from queue \n",ps, msg->pos); +#endif /* Calling transmit continuation */ if (( NULL != ps->pending_msgs_tail) && (NULL != ps->pending_msgs_tail->transmit_cont)) msg->transmit_cont (ps->pending_msgs_tail->transmit_cont_cls,&(ps->peercontext)->identity,GNUNET_OK); @@ -1174,13 +1137,17 @@ static size_t send_curl_read_callback(void *stream, size_t size, size_t nmemb, v * @param ptr destination pointer, passed to the libcurl handle * @return bytes read from stream */ -static size_t send_curl_write_callback( void *stream, size_t size, size_t nmemb, void *ptr) +static size_t curl_receive_cb( void *stream, size_t size, size_t nmemb, void *ptr) { - struct HTTP_Session * ps = ptr; + struct Session * ps = ptr; +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: %u bytes received\n",ps, size*nmemb); +#endif - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: %u bytes recieved\n",ps, size*nmemb); - GNUNET_SERVER_mst_receive(ps->msgtok, ps, stream, size*nmemb, GNUNET_NO, GNUNET_NO); + struct GNUNET_MessageHeader * msg = stream; + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: %u bytes msg->type %u msg->size %u\n",ps, size*nmemb, ntohs(msg->type), ntohs(msg->size)); + // GNUNET_SERVER_mst_receive(ps->msgtok, ps, stream, size*nmemb, GNUNET_NO, GNUNET_NO); return (size * nmemb); } @@ -1202,7 +1169,7 @@ static size_t send_schedule(void *cls, struct Session* ses ); * @param con connection * @return bytes sent to peer */ -static ssize_t send_check_connections (void *cls, struct Session* ses , struct HTTP_Session *ps) +static ssize_t send_check_connections (void *cls, struct Session *ps) { struct Plugin *plugin = cls; int bytes_sent = 0; @@ -1214,6 +1181,7 @@ static ssize_t send_check_connections (void *cls, struct Session* ses , struct H if (ps->direction == OUTBOUND) { + /* RECV DIRECTION */ /* Check if session is connected to receive data, otherwise connect to peer */ if (ps->recv_connected == GNUNET_NO) { @@ -1224,11 +1192,11 @@ static ssize_t send_check_connections (void *cls, struct Session* ses , struct H curl_easy_setopt(ps->recv_endpoint, CURLOPT_VERBOSE, 1L); #endif curl_easy_setopt(ps->recv_endpoint, CURLOPT_URL, ps->url); - curl_easy_setopt(ps->recv_endpoint, CURLOPT_HEADERFUNCTION, &curl_header_function); + curl_easy_setopt(ps->recv_endpoint, CURLOPT_HEADERFUNCTION, &curl_get_header_function); curl_easy_setopt(ps->recv_endpoint, CURLOPT_WRITEHEADER, ps); - curl_easy_setopt(ps->recv_endpoint, CURLOPT_READFUNCTION, send_curl_read_callback); + curl_easy_setopt(ps->recv_endpoint, CURLOPT_READFUNCTION, curl_send_cb); curl_easy_setopt(ps->recv_endpoint, CURLOPT_READDATA, ps); - curl_easy_setopt(ps->recv_endpoint, CURLOPT_WRITEFUNCTION, send_curl_write_callback); + curl_easy_setopt(ps->recv_endpoint, CURLOPT_WRITEFUNCTION, curl_receive_cb); curl_easy_setopt(ps->recv_endpoint, CURLOPT_WRITEDATA, ps); curl_easy_setopt(ps->recv_endpoint, CURLOPT_TIMEOUT, (long) timeout.value); curl_easy_setopt(ps->recv_endpoint, CURLOPT_PRIVATE, ps); @@ -1244,37 +1212,47 @@ static ssize_t send_check_connections (void *cls, struct Session* ses , struct H curl_multi_strerror (mret)); return -1; } - - if (ps->msgtok != NULL) - ps->msgtok = GNUNET_SERVER_mst_create (&curl_write_mst_cb, ps); + bytes_sent = send_schedule (plugin, NULL); +#if DEBUG_CONNECTIONS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: inbound not connected, initiating connection\n",ps); +#endif } } + /* waiting for receive direction */ + if (ps->recv_connected==GNUNET_NO) + return 0; + + /* SEND DIRECTION */ /* Check if session is connected to send data, otherwise connect to peer */ if ((ps->send_connected == GNUNET_YES) && (ps->send_endpoint!= NULL)) { - if (ps->send_connected == GNUNET_NO) + if (ps->send_active == GNUNET_YES) { +#if DEBUG_CONNECTIONS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: outbound active, enqueueing message\n",ps); +#endif return bytes_sent; } if (ps->send_active == GNUNET_NO) { +#if DEBUG_CONNECTIONS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: outbound paused, unpausing existing connection and enqueueing message\n",ps); +#endif curl_easy_pause(ps->send_endpoint,CURLPAUSE_CONT); ps->send_active=GNUNET_YES; return bytes_sent; } } - /* not connected, initiate connection */ - if ( NULL == ps->send_endpoint) + if ((ps->send_connected==GNUNET_NO) && (NULL == ps->send_endpoint)) ps->send_endpoint = curl_easy_init(); GNUNET_assert (ps->send_endpoint != NULL); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: outbound not connected, initiating connection\n",ps); - GNUNET_assert (NULL != ps->pending_msgs_tail); +#if DEBUG_CONNECTIONS + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Connection %X: outbound not connected, initiating connection\n",ps); +#endif + ps->send_active = GNUNET_NO; msg = ps->pending_msgs_tail; #if DEBUG_CURL @@ -1282,9 +1260,11 @@ static ssize_t send_check_connections (void *cls, struct Session* ses , struct H #endif curl_easy_setopt(ps->send_endpoint, CURLOPT_URL, ps->url); curl_easy_setopt(ps->send_endpoint, CURLOPT_PUT, 1L); - curl_easy_setopt(ps->send_endpoint, CURLOPT_READFUNCTION, send_curl_read_callback); + curl_easy_setopt(ps->send_endpoint, CURLOPT_HEADERFUNCTION, &curl_put_header_function); + curl_easy_setopt(ps->send_endpoint, CURLOPT_WRITEHEADER, ps); + curl_easy_setopt(ps->send_endpoint, CURLOPT_READFUNCTION, curl_send_cb); curl_easy_setopt(ps->send_endpoint, CURLOPT_READDATA, ps); - curl_easy_setopt(ps->send_endpoint, CURLOPT_WRITEFUNCTION, send_curl_write_callback); + curl_easy_setopt(ps->send_endpoint, CURLOPT_WRITEFUNCTION, curl_receive_cb); curl_easy_setopt(ps->send_endpoint, CURLOPT_READDATA, ps); curl_easy_setopt(ps->send_endpoint, CURLOPT_TIMEOUT, (long) timeout.value); curl_easy_setopt(ps->send_endpoint, CURLOPT_PRIVATE, ps); @@ -1300,18 +1280,19 @@ static ssize_t send_check_connections (void *cls, struct Session* ses , struct H curl_multi_strerror (mret)); return -1; } - ps->send_connected = GNUNET_YES; bytes_sent = send_schedule (plugin, NULL); return bytes_sent; } if (ps->direction == INBOUND) { GNUNET_assert (NULL != ps->pending_msgs_tail); + bytes_sent = 0; msg = ps->pending_msgs_tail; - if ((ps->recv_connected==GNUNET_YES) && (ps->recv_connected==GNUNET_YES)) + if ((ps->recv_connected==GNUNET_YES) && (ps->send_connected==GNUNET_YES)) bytes_sent = msg->size; return bytes_sent; } + return 0; } static void send_execute (void *cls, @@ -1322,8 +1303,9 @@ static void send_execute (void *cls, int running; struct CURLMsg *msg; CURLMcode mret; - struct HTTP_Session *ps = NULL; + struct Session *ps = NULL; struct HTTP_PeerContext *pc = NULL; + struct HTTP_Message * cur_msg = NULL; long http_result; GNUNET_assert(cls !=NULL); @@ -1359,6 +1341,7 @@ static void send_execute (void *cls, /* sending msg failed*/ if (msg->easy_handle == ps->send_endpoint) { +#if DEBUG_CONNECTIONS GNUNET_log(GNUNET_ERROR_TYPE_INFO, _("Connection %X: HTTP PUT to peer `%s' (`%s') failed: `%s' `%s'\n"), ps, @@ -1366,16 +1349,20 @@ static void send_execute (void *cls, http_plugin_address_to_string(NULL, ps->addr, ps->addrlen), "curl_multi_perform", curl_easy_strerror (msg->data.result)); - +#endif ps->send_connected = GNUNET_NO; + ps->send_active = GNUNET_NO; + curl_multi_remove_handle(plugin->multi_handle,msg->easy_handle); curl_easy_cleanup(ps->send_endpoint); ps->send_endpoint=NULL; - if (( NULL != ps->pending_msgs_tail) && ( NULL != ps->pending_msgs_tail->transmit_cont)) - ps->pending_msgs_tail->transmit_cont (ps->pending_msgs_tail->transmit_cont_cls,&pc->identity,GNUNET_SYSERR); + cur_msg = ps->pending_msgs_tail; + if (( NULL != cur_msg) && ( NULL != cur_msg->transmit_cont)) + cur_msg->transmit_cont (cur_msg->transmit_cont_cls,&pc->identity,GNUNET_SYSERR); } /* GET connection failed */ if (msg->easy_handle == ps->recv_endpoint) { +#if DEBUG_CONNECTIONS GNUNET_log(GNUNET_ERROR_TYPE_INFO, _("Connection %X: HTTP GET to peer `%s' (`%s') failed: `%s' `%s'\n"), ps, @@ -1383,7 +1370,10 @@ static void send_execute (void *cls, http_plugin_address_to_string(NULL, ps->addr, ps->addrlen), "curl_multi_perform", curl_easy_strerror (msg->data.result)); +#endif ps->recv_connected = GNUNET_NO; + ps->recv_active = GNUNET_NO; + curl_multi_remove_handle(plugin->multi_handle,msg->easy_handle); curl_easy_cleanup(ps->recv_endpoint); ps->recv_endpoint=NULL; } @@ -1393,49 +1383,53 @@ static void send_execute (void *cls, if (msg->easy_handle == ps->send_endpoint) { GNUNET_assert (CURLE_OK == curl_easy_getinfo(msg->easy_handle, CURLINFO_RESPONSE_CODE, &http_result)); +#if DEBUG_CONNECTIONS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connection %X: HTTP PUT connection to peer `%s' (`%s') was closed with HTTP code %u\n", ps, GNUNET_i2s(&pc->identity), http_plugin_address_to_string(NULL, ps->addr, ps->addrlen), http_result); - +#endif /* Calling transmit continuation */ - if (( NULL != ps->pending_msgs_tail) && (NULL != ps->pending_msgs_tail->transmit_cont)) + cur_msg = ps->pending_msgs_tail; + if (( NULL != cur_msg) && (NULL != cur_msg->transmit_cont)) { /* HTTP 1xx : Last message before here was informational */ if ((http_result >=100) && (http_result < 200)) - ps->pending_msgs_tail->transmit_cont (ps->pending_msgs_tail->transmit_cont_cls,&pc->identity,GNUNET_OK); + cur_msg->transmit_cont (cur_msg->transmit_cont_cls,&pc->identity,GNUNET_OK); /* HTTP 2xx: successful operations */ if ((http_result >=200) && (http_result < 300)) - ps->pending_msgs_tail->transmit_cont (ps->pending_msgs_tail->transmit_cont_cls,&pc->identity,GNUNET_OK); + cur_msg->transmit_cont (cur_msg->transmit_cont_cls,&pc->identity,GNUNET_OK); /* HTTP 3xx..5xx: error */ if ((http_result >=300) && (http_result < 600)) - ps->pending_msgs_tail->transmit_cont (ps->pending_msgs_tail->transmit_cont_cls,&pc->identity,GNUNET_SYSERR); + cur_msg->transmit_cont (cur_msg->transmit_cont_cls,&pc->identity,GNUNET_SYSERR); } ps->send_connected = GNUNET_NO; + ps->send_active = GNUNET_NO; + curl_multi_remove_handle(plugin->multi_handle,msg->easy_handle); curl_easy_cleanup(ps->send_endpoint); ps->send_endpoint =NULL; } if (msg->easy_handle == ps->recv_endpoint) { +#if DEBUG_CONNECTIONS GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Connection %X: HTTP GET connection to peer `%s' (`%s') was closed with HTTP code %u\n", ps, GNUNET_i2s(&pc->identity), http_plugin_address_to_string(NULL, ps->addr, ps->addrlen), http_result); - +#endif ps->recv_connected = GNUNET_NO; + ps->recv_active = GNUNET_NO; + curl_multi_remove_handle(plugin->multi_handle,msg->easy_handle); curl_easy_cleanup(ps->recv_endpoint); ps->recv_endpoint=NULL; } } - if (ps->pending_msgs_tail != NULL) - { - if (ps->pending_msgs_tail->pos>0) - remove_http_message(ps, ps->pending_msgs_tail); - } + if ((ps->recv_connected == GNUNET_NO) && (ps->send_connected == GNUNET_NO)) + remove_session (pc, ps, GNUNET_YES, GNUNET_SYSERR); return; default: break; @@ -1563,101 +1557,129 @@ http_plugin_send (void *cls, void *cont_cls) { struct Plugin *plugin = cls; - //struct Session *cs; struct HTTP_Message *msg; - //struct HTTP_Connection *con; - struct HTTP_PeerContext * pc; - struct HTTP_Session * ps; + struct Session * ps = NULL; + struct Session * ps_tmp = NULL; GNUNET_assert(cls !=NULL); - GNUNET_assert ((addr!=NULL) && (addrlen != 0)); - - /* get session from hashmap */ - //cs = session_get(plugin, target); - //con = session_check_outbound_address(plugin, cs, addr, addrlen); + char * force = GNUNET_malloc(40); + if (force_address == GNUNET_YES) + strcpy(force,"forced addr."); + if (force_address == GNUNET_NO) + strcpy(force,"any addr."); + if (force_address == GNUNET_SYSERR) + strcpy(force,"reliable bi-direc. address addr."); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Transport tells me to send %u bytes to `%s' using %s (%s) and session: %X\n", + msgbuf_size, + GNUNET_i2s(target), + force, + http_plugin_address_to_string(NULL, addr, addrlen), + session); + GNUNET_free(force); pc = GNUNET_CONTAINER_multihashmap_get (plugin->peers, &target->hashPubKey); /* Peer unknown */ if (pc==NULL) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"SEND: CREATING NEW PEER CONTEXT\n"); pc = GNUNET_malloc(sizeof (struct HTTP_PeerContext)); + pc->plugin = plugin; + pc->session_id_counter=1; memcpy(&pc->identity, target, sizeof(struct GNUNET_PeerIdentity)); GNUNET_CONTAINER_multihashmap_put(plugin->peers, &pc->identity.hashPubKey, pc, GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY); } - else + + /* Search for existing session using the passed address */ + if ((addr!=NULL) && (addrlen != 0)) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"SEND: PEER CONTEXT FOUND\n"); + ps = get_Session(plugin, pc, addr, addrlen); } - ps = get_HTTP_Session(plugin, pc, addr, addrlen); - /* session not existing, but address forced -> creating new session */ - if ((ps==NULL) && (force_address == GNUNET_YES)) + if (ps != NULL) + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Found existing connection to peer %s with given address, using %X\n", GNUNET_i2s(target), ps); + + /* Search for existing session using the passed session */ + if ((ps==NULL) && (force_address != GNUNET_YES)) { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"SEND: CREATING NEW SESSION %s\n",http_plugin_address_to_string(NULL, addr, addrlen)); - ps = GNUNET_malloc(sizeof (struct HTTP_Session)); - ps->addr = GNUNET_malloc(addrlen); - memcpy(ps->addr,addr,addrlen); - ps->addrlen = addrlen; - ps->direction=OUTBOUND; - ps->recv_connected = GNUNET_NO; - ps->send_connected = GNUNET_NO; - ps->pending_msgs_head = NULL; - ps->pending_msgs_tail = NULL; - ps->url = create_url (plugin, ps->addr, ps->addrlen); - GNUNET_CONTAINER_DLL_insert(pc->head,pc->tail,ps); + ps_tmp = pc->head; + while (ps_tmp!=NULL) + { + if ((ps_tmp==session) && (ps_tmp->recv_force_disconnect==GNUNET_NO) && (ps_tmp->send_force_disconnect==GNUNET_NO) && + (ps_tmp->recv_connected==GNUNET_YES) && (ps_tmp->send_connected==GNUNET_YES)) + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Found existing connection to peer %s with given session, using inbound session %X\n", GNUNET_i2s(target), ps_tmp); + ps = ps_tmp; + break; + } + ps_tmp=ps_tmp->next; + } } + /* session not existing, address not forced -> looking for other session */ - if ((ps==NULL) && (force_address == GNUNET_NO)) + if ((ps==NULL) && (force_address != GNUNET_YES)) { - /* FIXME: CREATING SESSION, SHOULD CHOOSE EXISTING */ - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"SEND: CREATING NEW SESSION %s\n",http_plugin_address_to_string(NULL, addr, addrlen)); - ps = GNUNET_malloc(sizeof (struct HTTP_Session)); - ps->addr = GNUNET_malloc(addrlen); - memcpy(ps->addr,addr,addrlen); - ps->addrlen = addrlen; - ps->direction=OUTBOUND; - ps->recv_connected = GNUNET_NO; - ps->send_connected = GNUNET_NO; - ps->pending_msgs_head = NULL; - ps->pending_msgs_tail = NULL; - ps->url = create_url (plugin, ps->addr, ps->addrlen); - GNUNET_CONTAINER_DLL_insert(pc->head,pc->tail,ps); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"No existing connection, but free to choose existing, searching for existing connection to peer %s\n", GNUNET_i2s(target)); + /* Choosing different session to peer when possible */ + struct Session * tmp = pc->head; + while (tmp!=NULL) + { + if ((tmp->recv_connected) && (tmp->send_connected) && (tmp->recv_force_disconnect==GNUNET_NO) && (tmp->send_force_disconnect==GNUNET_NO)) + { + ps = tmp; + } + tmp = tmp->next; + } + if (ps != NULL) + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"No existing connection to peer %s, selected connection %X\n", GNUNET_i2s(target),ps); + else + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"No existing connection to peer %s, no connection found\n", GNUNET_i2s(target)); } - if ((ps==NULL) && (force_address == GNUNET_SYSERR)) + + /* session not existing, but address forced -> creating new session */ + if ((ps==NULL) || ((ps==NULL) && (force_address == GNUNET_YES))) { - /* FIXME: CREATING SESSION, SHOULD CHOOSE EXISTING */ - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"SEND: CREATING NEW SESSION %s\n",http_plugin_address_to_string(NULL, addr, addrlen)); - ps = GNUNET_malloc(sizeof (struct HTTP_Session)); - ps->addr = GNUNET_malloc(addrlen); - memcpy(ps->addr,addr,addrlen); - ps->addrlen = addrlen; - ps->direction=OUTBOUND; - ps->recv_connected = GNUNET_NO; - ps->send_connected = GNUNET_NO; - ps->pending_msgs_head = NULL; - ps->pending_msgs_tail = NULL; - ps->url = create_url (plugin, ps->addr, ps->addrlen); - GNUNET_CONTAINER_DLL_insert(pc->head,pc->tail,ps); - } + if ((addr!=NULL) && (addrlen!=0)) + { + if (force_address == GNUNET_YES) + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"No existing connection & forced address: creating new connection to peer %s\n", GNUNET_i2s(target)); + if (force_address != GNUNET_YES) + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"No existing connection: creating new connection to peer %s\n", GNUNET_i2s(target)); - char * force = GNUNET_malloc(30); - if (force_address == GNUNET_YES) - strcpy(force,"forced addr."); - if (force_address == GNUNET_NO) - strcpy(force,"any addr."); - if (force_address == GNUNET_SYSERR) - strcpy(force,"reliable bi-direc. address addr."); - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Transport tells me to send %u bytes to `%s' %s (%s), session: %X\n", - msgbuf_size, - GNUNET_i2s(&pc->identity), - force, - http_plugin_address_to_string(NULL, addr, addrlen), - ps); + ps = GNUNET_malloc(sizeof (struct Session)); + if ((addrlen!=0) && (addr!=NULL)) + { + ps->addr = GNUNET_malloc(addrlen); + memcpy(ps->addr,addr,addrlen); + ps->addrlen = addrlen; + } + else + { + ps->addr = NULL; + ps->addrlen = 0; + } + ps->direction=OUTBOUND; + ps->recv_connected = GNUNET_NO; + ps->recv_force_disconnect = GNUNET_NO; + ps->send_connected = GNUNET_NO; + ps->send_force_disconnect = GNUNET_NO; + ps->pending_msgs_head = NULL; + ps->pending_msgs_tail = NULL; + ps->peercontext=pc; + ps->session_id = pc->session_id_counter; + pc->session_id_counter++; + ps->url = create_url (plugin, ps->addr, ps->addrlen, ps->session_id); + if (ps->msgtok == NULL) + ps->msgtok = GNUNET_SERVER_mst_create (&curl_receive_mst_cb, ps); + GNUNET_CONTAINER_DLL_insert(pc->head,pc->tail,ps); + } + else + { + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"No existing session & and no address given: no way to send this message to peer `%s'!\n", GNUNET_i2s(target)); + return -1; + } + } - GNUNET_free(force); /* create msg */ msg = GNUNET_malloc (sizeof (struct HTTP_Message) + msgbuf_size); msg->next = NULL; @@ -1669,7 +1691,7 @@ http_plugin_send (void *cls, memcpy (msg->buf,msgbuf, msgbuf_size); GNUNET_CONTAINER_DLL_insert(ps->pending_msgs_head,ps->pending_msgs_tail,msg); - return send_check_connections (plugin, session, ps); + return send_check_connections (plugin, ps); } @@ -1686,26 +1708,50 @@ static void http_plugin_disconnect (void *cls, const struct GNUNET_PeerIdentity *target) { + struct Plugin *plugin = cls; - struct HTTP_Connection *con; - struct Session *cs; + struct HTTP_PeerContext *pc = NULL; + struct Session *ps = NULL; - /* get session from hashmap */ - cs = session_get(plugin, target); - con = cs->outbound_connections_head; + pc = GNUNET_CONTAINER_multihashmap_get (plugin->peers, &target->hashPubKey); + if (pc==NULL) + return; + + ps = pc->head; - while (con!=NULL) + while (ps!=NULL) { - if (con->put_curl_handle!=NULL) - curl_easy_cleanup(con->put_curl_handle); - con->put_curl_handle=NULL; - con->put_connected = GNUNET_NO; - while (con->pending_msgs_head!=NULL) + if (ps->direction==OUTBOUND) { - remove_http_message(con, con->pending_msgs_head); + if (ps->send_endpoint!=NULL) + { + curl_multi_remove_handle(plugin->multi_handle,ps->send_endpoint); + curl_easy_cleanup(ps->send_endpoint); + ps->send_endpoint=NULL; + ps->send_force_disconnect = GNUNET_YES; + } + if (ps->recv_endpoint!=NULL) + { + curl_multi_remove_handle(plugin->multi_handle,ps->recv_endpoint); + curl_easy_cleanup(ps->recv_endpoint); + ps->recv_endpoint=NULL; + ps->recv_force_disconnect = GNUNET_YES; + } } - con=con->next; + if (ps->direction==INBOUND) + { + ps->recv_force_disconnect = GNUNET_YES; + ps->send_force_disconnect = GNUNET_YES; + } + while (ps->pending_msgs_head!=NULL) + { + remove_http_message(ps, ps->pending_msgs_head); + } + ps->recv_active = GNUNET_NO; + ps->send_active = GNUNET_NO; + ps=ps->next; } + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"All connections to peer `%s' terminated\n", GNUNET_i2s(target)); } @@ -1942,40 +1988,49 @@ process_interfaces (void *cls, } return GNUNET_OK; } -int hashMapFreeIterator (void *cls, const GNUNET_HashCode *key, void *value) + +int remove_peer_context_Iterator (void *cls, const GNUNET_HashCode *key, void *value) { - struct Session * cs = value; - struct HTTP_Connection * con = cs->outbound_connections_head; - struct HTTP_Connection * tmp_con = cs->outbound_connections_head; + struct HTTP_PeerContext * pc = value; + struct Session * ps = pc->head; + struct Session * tmp = NULL; struct HTTP_Message * msg = NULL; - struct HTTP_Message * tmp_msg = NULL; + struct HTTP_Message * msg_tmp = NULL; - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Freeing session for peer `%s'\n",GNUNET_i2s(&cs->identity)); + GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"Freeing context for peer `%s'\n",GNUNET_i2s(&pc->identity)); - /* freeing connections */ - while (con!=NULL) + while (ps!=NULL) { - GNUNET_free(con->url); - if (con->put_curl_handle!=NULL) - curl_easy_cleanup(con->put_curl_handle); - con->put_curl_handle = NULL; - msg = con->pending_msgs_head; + tmp = ps->next; + + GNUNET_free_non_null (ps->addr); + GNUNET_free(ps->url); + if (ps->msgtok != NULL) + GNUNET_SERVER_mst_destroy (ps->msgtok); + + msg = ps->pending_msgs_head; while (msg!=NULL) { - tmp_msg=msg->next; + msg_tmp = msg->next; GNUNET_free(msg); - msg = tmp_msg; + msg = msg_tmp; + } + if (ps->direction==OUTBOUND) + { + if (ps->send_endpoint!=NULL) + curl_easy_cleanup(ps->send_endpoint); + if (ps->recv_endpoint!=NULL) + curl_easy_cleanup(ps->recv_endpoint); } - tmp_con=con->next; - GNUNET_free(con); - con=tmp_con; - } - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,"All sessions freed \n"); - GNUNET_free (cs); + GNUNET_free(ps); + ps=tmp; + } + GNUNET_free(pc); return GNUNET_YES; } + /** * Exit point from the plugin. */ @@ -1988,7 +2043,6 @@ libgnunet_plugin_transport_http_done (void *cls) GNUNET_assert(cls !=NULL); - if ( plugin->http_server_task_v4 != GNUNET_SCHEDULER_NO_TASK) { GNUNET_SCHEDULER_cancel(plugin->env->sched, plugin->http_server_task_v4); @@ -2018,12 +2072,10 @@ libgnunet_plugin_transport_http_done (void *cls) plugin->http_server_daemon_v6 = NULL; } - /* free all sessions */ - GNUNET_CONTAINER_multihashmap_iterate (plugin->sessions, - &hashMapFreeIterator, + /* free all peer information */ + GNUNET_CONTAINER_multihashmap_iterate (plugin->peers, + &remove_peer_context_Iterator, NULL); - - GNUNET_CONTAINER_multihashmap_destroy (plugin->sessions); GNUNET_CONTAINER_multihashmap_destroy (plugin->peers); mret = curl_multi_cleanup(plugin->multi_handle); @@ -2054,7 +2106,7 @@ libgnunet_plugin_transport_http_init (void *cls) plugin = GNUNET_malloc (sizeof (struct Plugin)); plugin->env = env; - plugin->sessions = NULL; + plugin->peers = NULL; api = GNUNET_malloc (sizeof (struct GNUNET_TRANSPORT_PluginFunctions)); api->cls = plugin; @@ -2090,23 +2142,23 @@ libgnunet_plugin_transport_http_init (void *cls) { plugin->http_server_daemon_v6 = MHD_start_daemon (MHD_USE_IPv6, port, - &acceptPolicyCallback, - plugin , &accessHandlerCallback, plugin, + &mhd_accept_cb, + plugin , &mdh_access_cb, plugin, MHD_OPTION_CONNECTION_LIMIT, (unsigned int) 16, MHD_OPTION_PER_IP_CONNECTION_LIMIT, (unsigned int) 1, MHD_OPTION_CONNECTION_TIMEOUT, (gn_timeout.value / 1000), MHD_OPTION_CONNECTION_MEMORY_LIMIT, (size_t) (16 * 1024), - MHD_OPTION_NOTIFY_COMPLETED, &requestCompletedCallback, NULL, + MHD_OPTION_NOTIFY_COMPLETED, &mhd_termination_cb, NULL, MHD_OPTION_END); plugin->http_server_daemon_v4 = MHD_start_daemon (MHD_NO_FLAG, port, - &acceptPolicyCallback, - plugin , &accessHandlerCallback, plugin, + &mhd_accept_cb, + plugin , &mdh_access_cb, plugin, MHD_OPTION_CONNECTION_LIMIT, (unsigned int) 16, MHD_OPTION_PER_IP_CONNECTION_LIMIT, (unsigned int) 1, MHD_OPTION_CONNECTION_TIMEOUT, (gn_timeout.value / 1000), MHD_OPTION_CONNECTION_MEMORY_LIMIT, (size_t) (16 * 1024), - MHD_OPTION_NOTIFY_COMPLETED, &requestCompletedCallback, NULL, + MHD_OPTION_NOTIFY_COMPLETED, &mhd_termination_cb, NULL, MHD_OPTION_END); } if (plugin->http_server_daemon_v4 != NULL) @@ -2139,7 +2191,6 @@ libgnunet_plugin_transport_http_init (void *cls) return NULL; } - plugin->sessions = GNUNET_CONTAINER_multihashmap_create (10); plugin->peers = GNUNET_CONTAINER_multihashmap_create (10); GNUNET_OS_network_interfaces_list (&process_interfaces, plugin);