/* vim: set expandtab ts=4 sw=4: */ /* * You may redistribute this program and/or modify it under the terms of * the GNU General Public License as published by the Free Software Foundation, * either version 3 of the License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ #include "benc/String.h" #include "net/SwitchPinger.h" #include "dht/Address.h" #include "util/Bits.h" #include "util/Checksum.h" #include "util/Endian.h" #include "util/Pinger.h" #include "util/version/Version.h" #include "util/Identity.h" #include "wire/Headers.h" #include "wire/Control.h" #include "wire/Error.h" struct SwitchPinger { struct Interface* iface; struct Pinger* pinger; struct Admin* admin; struct Log* logger; struct Allocator* allocator; /** * The label is stored here while the message is sent through the pinger * and it decides which ping the incoming message belongs to. */ uint64_t incomingLabel; /** The version of the node which sent the message. */ uint32_t incomingVersion; bool isError; /** Pings which are currently waiting for responses. */ int outstandingPings; /** Maximum number of pings which can be outstanding at one time. */ int maxConcurrentPings; Identity }; struct Ping { struct SwitchPinger_Ping public; uint64_t label; String* data; struct SwitchPinger* context; SwitchPinger_ResponseCallback onResponse; void* onResponseContext; struct Pinger_Ping* pingerPing; Identity }; // incoming message from network, pointing to the beginning of the switch header. static uint8_t receiveMessage(struct Message* msg, struct Interface* iface) { struct SwitchPinger* ctx = Identity_cast((struct SwitchPinger*) iface->receiverContext); struct Headers_SwitchHeader* switchHeader = (struct Headers_SwitchHeader*) msg->bytes; ctx->incomingLabel = Endian_bigEndianToHost64(switchHeader->label_be); ctx->incomingVersion = 0; Assert_true(Headers_getMessageType(switchHeader) == Headers_SwitchHeader_TYPE_CONTROL); Message_shift(msg, -Headers_SwitchHeader_SIZE, NULL); struct Control* ctrl = (struct Control*) msg->bytes; if (ctrl->type_be == Control_PONG_be) { Message_shift(msg, -Control_HEADER_SIZE, NULL); ctx->isError = false; struct Control_Ping* pongHeader = (struct Control_Ping*) msg->bytes; if (msg->length >= Control_Pong_MIN_SIZE) { ctx->incomingVersion = Endian_bigEndianToHost32(pongHeader->version_be); Message_shift(msg, -Control_Pong_HEADER_SIZE, NULL); if (pongHeader->magic != Control_Pong_MAGIC) { Log_debug(ctx->logger, "dropped invalid switch pong"); return Error_INVALID; } } else { Log_debug(ctx->logger, "got runt pong message, length: [%d]", msg->length); return Error_INVALID; } } else if (ctrl->type_be == Control_ERROR_be) { Log_debug(ctx->logger, "error was caused by our ping."); Message_shift(msg, -( Control_HEADER_SIZE + Control_Error_HEADER_SIZE + Headers_SwitchHeader_SIZE + Control_HEADER_SIZE + Control_Ping_HEADER_SIZE ), NULL); ctx->isError = true; } else { // If it gets here then Ducttape.c is failing. Assert_always(false); } String* msgStr = &(String) { .bytes = (char*) msg->bytes, .len = msg->length }; Pinger_pongReceived(msgStr, ctx->pinger); return Error_NONE; } static void onPingResponse(String* data, uint32_t milliseconds, void* vping) { struct Ping* p = Identity_cast((struct Ping*) vping); enum SwitchPinger_Result err = SwitchPinger_Result_OK; uint64_t label = p->context->incomingLabel; if (data) { if (label != p->label) { err = SwitchPinger_Result_LABEL_MISMATCH; } else if ((p->data || data->len > 0) && !String_equals(data, p->data)) { err = SwitchPinger_Result_WRONG_DATA; } else if (p->context->isError) { err = SwitchPinger_Result_ERROR_RESPONSE; } } else { err = SwitchPinger_Result_TIMEOUT; } uint32_t version = p->context->incomingVersion; p->onResponse(err, label, data, milliseconds, version, p->public.onResponseContext); } static void sendPing(String* data, void* sendPingContext) { struct Ping* p = Identity_cast((struct Ping*) sendPingContext); #define BUFFER_SZ 4096 uint8_t buffer[BUFFER_SZ]; struct Message msg = { .length = data->len, // Make it aligned along an 8 byte boundry (assuming the buffer is) .bytes = &buffer[BUFFER_SZ - (data->len + 8 - (data->len % 8))] }; msg.padding = msg.bytes - buffer; Assert_true(data->len < (BUFFER_SZ / 2)); Bits_memcpy(msg.bytes, data->bytes, data->len); Message_shift(&msg, Control_Ping_HEADER_SIZE, NULL); struct Control_Ping* pingHeader = (struct Control_Ping*) msg.bytes; pingHeader->magic = Control_Ping_MAGIC; pingHeader->version_be = Endian_hostToBigEndian32(Version_CURRENT_PROTOCOL); Message_shift(&msg, Control_HEADER_SIZE, NULL); struct Control* ctrl = (struct Control*) msg.bytes; ctrl->checksum_be = 0; ctrl->type_be = Control_PING_be; ctrl->checksum_be = Checksum_engine(msg.bytes, msg.length); Message_shift(&msg, Headers_SwitchHeader_SIZE, NULL); struct Headers_SwitchHeader* switchHeader = (struct Headers_SwitchHeader*) msg.bytes; switchHeader->label_be = Endian_hostToBigEndian64(p->label); Headers_setPriorityAndMessageType(switchHeader, 0, Headers_SwitchHeader_TYPE_CONTROL); p->context->iface->sendMessage(&msg, p->context->iface); } #define BSTR_SIZEOF(x) &(String) { .bytes = x, .len = sizeof(x) - 1 } static String* RESULT_STRING_OK = BSTR_SIZEOF("pong"); static String* RESULT_STRING_LABEL_MISMATCH = BSTR_SIZEOF("pong has different label"); static String* RESULT_STRING_WRONG_DATA = BSTR_SIZEOF("data is different"); static String* RESULT_STRING_ERROR_RESPONSE = BSTR_SIZEOF("ping message caused switch error"); static String* RESULT_STRING_TIMEOUT = BSTR_SIZEOF("timeout"); static String* RESULT_STRING_UNKNOWN = BSTR_SIZEOF("unknown error"); String* SwitchPinger_resultString(enum SwitchPinger_Result result) { switch (result) { case SwitchPinger_Result_OK: return RESULT_STRING_OK; case SwitchPinger_Result_LABEL_MISMATCH: return RESULT_STRING_LABEL_MISMATCH; case SwitchPinger_Result_WRONG_DATA: return RESULT_STRING_WRONG_DATA; case SwitchPinger_Result_ERROR_RESPONSE: return RESULT_STRING_ERROR_RESPONSE; case SwitchPinger_Result_TIMEOUT: return RESULT_STRING_TIMEOUT; default: return RESULT_STRING_UNKNOWN; }; } static int onPingFree(struct Allocator_OnFreeJob* job) { struct Ping* ping = Identity_cast((struct Ping*)job->userData); struct SwitchPinger* ctx = Identity_cast(ping->context); ctx->outstandingPings--; Assert_true(ctx->outstandingPings >= 0); return 0; } struct SwitchPinger_Ping* SwitchPinger_newPing(uint64_t label, String* data, uint32_t timeoutMilliseconds, SwitchPinger_ResponseCallback onResponse, struct Allocator* alloc, struct SwitchPinger* ctx) { if (data && data->len > Control_Ping_MAX_SIZE) { return NULL; } if (ctx->outstandingPings > ctx->maxConcurrentPings) { Log_debug(ctx->logger, "Skipping switch ping because there are already [%d] outstanding", ctx->outstandingPings); return NULL; } struct Pinger_Ping* pp = Pinger_newPing(data, onPingResponse, sendPing, timeoutMilliseconds, alloc, ctx->pinger); struct Ping* ping = Allocator_clone(pp->pingAlloc, (&(struct Ping) { .public = { .pingAlloc = pp->pingAlloc }, .label = label, .data = String_clone(data, pp->pingAlloc), .context = ctx, .onResponse = onResponse, .pingerPing = pp })); Identity_set(ping); Allocator_onFree(pp->pingAlloc, onPingFree, ping); pp->context = ping; ctx->outstandingPings++; return &ping->public; } void SwitchPinger_sendPing(struct SwitchPinger_Ping* ping) { struct Ping* p = Identity_cast((struct Ping*) ping); Pinger_sendPing(p->pingerPing); } struct SwitchPinger* SwitchPinger_new(struct Interface* iface, struct EventBase* eventBase, struct Random* rand, struct Log* logger, struct Allocator* alloc) { struct SwitchPinger* sp = Allocator_malloc(alloc, sizeof(struct SwitchPinger)); Bits_memcpyConst(sp, (&(struct SwitchPinger) { .iface = iface, .pinger = Pinger_new(eventBase, rand, logger, alloc), .logger = logger, .allocator = alloc, .maxConcurrentPings = SwitchPinger_DEFAULT_MAX_CONCURRENT_PINGS, }), sizeof(struct SwitchPinger)); iface->receiveMessage = receiveMessage; iface->receiverContext = sp; Identity_set(sp); return sp; }