1 /*********************************************************************************************************
2 * Software License Agreement (BSD License) *
3 * Author: Sebastien Decugis <sdecugis@freediameter.net> *
4 * *
5 * Copyright (c) 2016, WIDE Project and NICT *
6 * All rights reserved. *
7 * *
8 * Redistribution and use of this software in source and binary forms, with or without modification, are *
9 * permitted provided that the following conditions are met: *
10 * *
11 * * Redistributions of source code must retain the above *
12 * copyright notice, this list of conditions and the *
13 * following disclaimer. *
14 * *
15 * * Redistributions in binary form must reproduce the above *
16 * copyright notice, this list of conditions and the *
17 * following disclaimer in the documentation and/or other *
18 * materials provided with the distribution. *
19 * *
20 * * Neither the name of the WIDE Project or NICT nor the *
21 * names of its contributors may be used to endorse or *
22 * promote products derived from this software without *
23 * specific prior written permission of WIDE Project and *
24 * NICT. *
25 * *
26 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED *
27 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A *
28 * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR *
29 * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT *
30 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS *
31 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR *
32 * TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF *
33 * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *
34 *********************************************************************************************************/
35
36 #include "fdcore-internal.h"
37
38 /* Alloc a new hbh for requests, bufferize the message and send on the connection, save in sentreq if provided */
do_send(struct msg ** msg,struct cnxctx * cnx,uint32_t * hbh,struct fd_peer * peer)39 static int do_send(struct msg ** msg, struct cnxctx * cnx, uint32_t * hbh, struct fd_peer * peer)
40 {
41 struct msg_hdr * hdr;
42 int msg_is_a_req;
43 uint8_t * buf;
44 size_t sz;
45 int ret;
46 uint32_t bkp_hbh = 0;
47 struct msg *cpy_for_logs_only;
48
49 TRACE_ENTRY("%p %p %p %p", msg, cnx, hbh, peer);
50
51 /* Retrieve the message header */
52 CHECK_FCT( fd_msg_hdr(*msg, &hdr) );
53
54 msg_is_a_req = (hdr->msg_flags & CMD_FLAG_REQUEST);
55 if (msg_is_a_req) {
56 CHECK_PARAMS(hbh && peer);
57 /* Alloc the hop-by-hop id and increment the value for next message */
58 bkp_hbh = hdr->msg_hbhid;
59 hdr->msg_hbhid = *hbh;
60 *hbh = hdr->msg_hbhid + 1;
61 }
62
63 /* Create the message buffer */
64 CHECK_FCT(fd_msg_bufferize( *msg, &buf, &sz ));
65 pthread_cleanup_push( free, buf );
66
67 cpy_for_logs_only = *msg;
68
69 /* Save a request before sending so that there is no race condition with the answer */
70 if (msg_is_a_req) {
71 CHECK_FCT_DO( ret = fd_p_sr_store(&peer->p_sr, msg, &hdr->msg_hbhid, bkp_hbh), goto out );
72 }
73
74 /* Log the message */
75 fd_hook_call(HOOK_MESSAGE_SENT, cpy_for_logs_only, peer, NULL, fd_msg_pmdl_get(cpy_for_logs_only));
76
77 pthread_cleanup_push((void *)fd_msg_free, *msg /* might be NULL, no problem */);
78
79 /* Send the message */
80 CHECK_FCT_DO( ret = fd_cnx_send(cnx, buf, sz), );
81
82 pthread_cleanup_pop(0);
83
84 out:
85 ;
86 pthread_cleanup_pop(1);
87
88 if (ret)
89 return ret;
90
91 /* Free remaining messages (i.e. answers) */
92 if (*msg) {
93 CHECK_FCT( fd_msg_free(*msg) );
94 *msg = NULL;
95 }
96
97 return 0;
98 }
99
100 /* The code of the "out" thread */
out_thr(void * arg)101 static void * out_thr(void * arg)
102 {
103 struct fd_peer * peer = arg;
104 int stop = 0;
105 struct msg * msg;
106 ASSERT( CHECK_PEER(peer) );
107
108 /* Set the thread name */
109 {
110 char buf[48];
111 snprintf(buf, sizeof(buf), "OUT/%s", peer->p_hdr.info.pi_diamid);
112 fd_log_threadname ( buf );
113 }
114
115 /* Loop until cancelation */
116 while (!stop) {
117 int ret;
118
119 /* Retrieve next message to send */
120 CHECK_FCT_DO( fd_fifo_get(peer->p_tosend, &msg), goto error );
121
122 /* Send the message, log any error */
123 CHECK_FCT_DO( ret = do_send(&msg, peer->p_cnxctx, &peer->p_hbh, peer),
124 {
125 if (msg) {
126 char buf[256];
127 snprintf(buf, sizeof(buf), "Error while sending this message: %s", strerror(ret));
128 fd_hook_call(HOOK_MESSAGE_DROPPED, msg, NULL, buf, fd_msg_pmdl_get(msg));
129 fd_msg_free(msg);
130 }
131 stop = 1;
132 } );
133
134 }
135
136 /* If we're here it means there was an error on the socket. We need to continue to purge the fifo & until we are canceled */
137 CHECK_FCT_DO( fd_event_send(peer->p_events, FDEVP_CNX_ERROR, 0, NULL), /* What do we do if it fails? */ );
138
139 /* Requeue all routable messages in the global "out" queue, until we are canceled once the PSM deals with the CNX_ERROR sent above */
140 while ( fd_fifo_get(peer->p_tosend, &msg) == 0 ) {
141 if (fd_msg_is_routable(msg)) {
142 CHECK_FCT_DO(fd_fifo_post_noblock(peer->p_tofailover, (void *)&msg),
143 {
144 /* fallback: destroy the message */
145 fd_hook_call(HOOK_MESSAGE_DROPPED, msg, NULL, "Internal error: unable to requeue this message during failover process", fd_msg_pmdl_get(msg));
146 CHECK_FCT_DO(fd_msg_free(msg), /* What can we do more? */)
147 } );
148 } else {
149 /* Just free it */
150 /* fd_hook_call(HOOK_MESSAGE_DROPPED, m, NULL, "Non-routable message freed during handover", fd_msg_pmdl_get(m)); */
151 CHECK_FCT_DO(fd_msg_free(msg), /* What can we do more? */)
152 }
153 }
154
155 error:
156 /* It is not really a connection error, but the effect is the same, we are not able to send anymore message */
157 CHECK_FCT_DO( fd_event_send(peer->p_events, FDEVP_CNX_ERROR, 0, NULL), /* What do we do if it fails? */ );
158 return NULL;
159 }
160
161 /* Wrapper to sending a message either by out thread (peer in OPEN state) or directly; cnx or peer must be provided. Flags are valid only for direct sending, not through thread (unused) */
fd_out_send(struct msg ** msg,struct cnxctx * cnx,struct fd_peer * peer,int update_reqin_cnt)162 int fd_out_send(struct msg ** msg, struct cnxctx * cnx, struct fd_peer * peer, int update_reqin_cnt)
163 {
164 struct msg_hdr * hdr;
165
166 TRACE_ENTRY("%p %p %p", msg, cnx, peer);
167 CHECK_PARAMS( msg && *msg && (cnx || (peer && peer->p_cnxctx)));
168
169 fd_hook_call(HOOK_MESSAGE_SENDING, *msg, peer, NULL, fd_msg_pmdl_get(*msg));
170
171 if (update_reqin_cnt && peer) {
172 CHECK_FCT( fd_msg_hdr(*msg, &hdr) );
173 if (!(hdr->msg_flags & CMD_FLAG_REQUEST)) {
174 /* Update the count of pending answers to send */
175 CHECK_POSIX( pthread_mutex_lock(&peer->p_state_mtx) );
176 peer->p_reqin_count--;
177 CHECK_POSIX( pthread_mutex_unlock(&peer->p_state_mtx) );
178 }
179 }
180
181 if (fd_peer_getstate(peer) == STATE_OPEN) {
182 /* Normal case: just queue for the out thread to pick it up */
183 CHECK_FCT( fd_fifo_post(peer->p_tosend, msg) );
184
185 } else {
186 int ret;
187 uint32_t *hbh = NULL;
188
189 /* In other cases, the thread is not running, so we handle the sending directly */
190 if (peer)
191 hbh = &peer->p_hbh;
192
193 if (!cnx)
194 cnx = peer->p_cnxctx;
195
196 /* Do send the message */
197 CHECK_FCT_DO( ret = do_send(msg, cnx, hbh, peer),
198 {
199 if (msg) {
200 char buf[256];
201 snprintf(buf, sizeof(buf), "Error while sending this message: %s", strerror(ret));
202 fd_hook_call(HOOK_MESSAGE_DROPPED, *msg, NULL, buf, fd_msg_pmdl_get(*msg));
203 fd_msg_free(*msg);
204 *msg = NULL;
205 }
206 } );
207 }
208
209 return 0;
210 }
211
212 /* Start the "out" thread that picks messages in p_tosend and send them on p_cnxctx */
fd_out_start(struct fd_peer * peer)213 int fd_out_start(struct fd_peer * peer)
214 {
215 TRACE_ENTRY("%p", peer);
216 CHECK_PARAMS( CHECK_PEER(peer) && (peer->p_outthr == (pthread_t)NULL) );
217
218 CHECK_POSIX( pthread_create(&peer->p_outthr, NULL, out_thr, peer) );
219
220 CHECK_FCT( fd_cnx_unordered_delivery(peer->p_cnxctx, 1) );
221
222 return 0;
223 }
224
225 /* Stop that thread */
fd_out_stop(struct fd_peer * peer)226 int fd_out_stop(struct fd_peer * peer)
227 {
228 TRACE_ENTRY("%p", peer);
229 CHECK_PARAMS( CHECK_PEER(peer) );
230
231 CHECK_FCT( fd_cnx_unordered_delivery(peer->p_cnxctx, 0) );
232
233 CHECK_FCT( fd_thr_term(&peer->p_outthr) );
234
235 return 0;
236 }
237
238