More impl of UDPConnection

Add client initiate connection.
This commit is contained in:
Stephen Seo 2019-03-04 15:04:16 +09:00
parent 0f1e9c9aed
commit 4776a2f202
3 changed files with 115 additions and 18 deletions

View file

@ -37,6 +37,7 @@ static const char *UDPC_ERR_THREADFAIL_STR = "Failed to create thread";
#define UDPC_BAD_MODE_SEND_INTERVAL (1.0f/10.0f) #define UDPC_BAD_MODE_SEND_INTERVAL (1.0f/10.0f)
#define UDPC_TIMEOUT_SECONDS 10.0f #define UDPC_TIMEOUT_SECONDS 10.0f
#define UDPC_HEARTBEAT_PKT_INTERVAL (15.0f/100.0f) #define UDPC_HEARTBEAT_PKT_INTERVAL (15.0f/100.0f)
#define UDPC_INIT_PKT_INTERVAL 5.0f
#define UDPC_PKT_PROTOCOL_ID 1357924680 #define UDPC_PKT_PROTOCOL_ID 1357924680
#define UDPC_ID_CONNECT 0x80000000 #define UDPC_ID_CONNECT 0x80000000

View file

@ -190,16 +190,16 @@ void UDPC_INTERNAL_destroy_conMap(void *unused, uint32_t addr, char *data)
} }
UDPC_Deque_destroy(cd->sendPktQueue); UDPC_Deque_destroy(cd->sendPktQueue);
for(int x = 0; x * sizeof(UDPC_INTERNAL_PacketInfo) < cd->resendPktQueue->size; ++x) for(int x = 0; x * sizeof(UDPC_INTERNAL_PacketInfo) < cd->priorityPktQueue->size; ++x)
{ {
UDPC_INTERNAL_PacketInfo *pinfo = UDPC_Deque_index_ptr( UDPC_INTERNAL_PacketInfo *pinfo = UDPC_Deque_index_ptr(
cd->resendPktQueue, sizeof(UDPC_INTERNAL_PacketInfo), x); cd->priorityPktQueue, sizeof(UDPC_INTERNAL_PacketInfo), x);
if(pinfo->data) if(pinfo->data)
{ {
free(pinfo->data); free(pinfo->data);
} }
} }
UDPC_Deque_destroy(cd->resendPktQueue); UDPC_Deque_destroy(cd->priorityPktQueue);
} }
void UDPC_set_callback_connected( void UDPC_set_callback_connected(
@ -289,6 +289,40 @@ void UDPC_check_events(UDPC_Context *ctx)
} }
} }
void UDPC_client_initiate_connection(UDPC_Context *ctx, uint32_t addr, uint16_t port)
{
if((ctx->flags & 0x2) == 0 || UDPC_HashMap_has(ctx->conMap, addr) != 0)
{
// must be client or no already-existing connection to same address
return;
}
UDPC_INTERNAL_ConnectionData cd = {
0x9,
0,
0,
0,
0xFFFFFFFF,
0.0f,
30.0f,
0.0f,
0.0f,
addr,
port,
UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_SENT_PKTS_ALLOC_SIZE),
UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_SEND_PKTS_ALLOC_SIZE),
UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_RESEND_PKTS_ALLOC_SIZE),
{0, 0},
{0, 0},
0.0f
};
timespec_get(&cd.received, TIME_UTC);
// only set "received" to now, since "sent" will be set after sending packet
UDPC_HashMap_insert(ctx->conMap, addr, &cd);
}
int UDPC_queue_send(UDPC_Context *ctx, uint32_t addr, uint32_t isChecked, void *data, uint32_t size) int UDPC_queue_send(UDPC_Context *ctx, uint32_t addr, uint32_t isChecked, void *data, uint32_t size)
{ {
UDPC_INTERNAL_ConnectionData *cd = UDPC_HashMap_get(ctx->conMap, addr); UDPC_INTERNAL_ConnectionData *cd = UDPC_HashMap_get(ctx->conMap, addr);
@ -455,7 +489,8 @@ void UDPC_update(UDPC_Context *ctx)
if(isConnect != 0 && (ctx->flags & 0x40) != 0) if(isConnect != 0 && (ctx->flags & 0x40) != 0)
{ {
if(!UDPC_HashMap_get(ctx->conMap, receivedData.sin_addr.s_addr)) if((ctx->flags & 0x2) == 0
&& !UDPC_HashMap_get(ctx->conMap, receivedData.sin_addr.s_addr))
{ {
UDPC_INTERNAL_log(ctx, 2, "Establishing connection with %s port %d", UDPC_INTERNAL_log(ctx, 2, "Establishing connection with %s port %d",
UDPC_INTERNAL_atostr(ctx, receivedData.sin_addr.s_addr), UDPC_INTERNAL_atostr(ctx, receivedData.sin_addr.s_addr),
@ -475,12 +510,10 @@ void UDPC_update(UDPC_Context *ctx)
UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_SENT_PKTS_ALLOC_SIZE), UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_SENT_PKTS_ALLOC_SIZE),
UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_SEND_PKTS_ALLOC_SIZE), UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_SEND_PKTS_ALLOC_SIZE),
UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_RESEND_PKTS_ALLOC_SIZE), UDPC_Deque_init(sizeof(UDPC_INTERNAL_PacketInfo) * UDPC_RESEND_PKTS_ALLOC_SIZE),
{0, 0}, us.tsNow,
{0, 0}, us.tsNow,
0.0f 0.0f
}; };
newCD.received = us.tsNow;
newCD.sent = us.tsNow;
UDPC_HashMap_insert(ctx->conMap, newCD.addr, &newCD); UDPC_HashMap_insert(ctx->conMap, newCD.addr, &newCD);
if(UDPC_Deque_get_available(ctx->connectedEvents) == 0) if(UDPC_Deque_get_available(ctx->connectedEvents) == 0)
{ {
@ -494,6 +527,16 @@ void UDPC_update(UDPC_Context *ctx)
UDPC_Deque_push_back(ctx->connectedEvents, &receivedData.sin_addr.s_addr, 4); UDPC_Deque_push_back(ctx->connectedEvents, &receivedData.sin_addr.s_addr, 4);
} }
} }
else if((ctx->flags & 0x2) != 0)
{
UDPC_INTERNAL_ConnectionData *cd = UDPC_HashMap_get(ctx->conMap, receivedData.sin_addr.s_addr);
if(!cd) { return; }
cd->flags &= 0xFFFFFFF7;
cd->id = conID;
UDPC_INTERNAL_log(ctx, 2, "Got id %u from server %s", conID,
UDPC_INTERNAL_atostr(ctx, receivedData.sin_addr.s_addr));
}
return; return;
} }
else if(isPing != 0) else if(isPing != 0)
@ -722,7 +765,47 @@ void UDPC_INTERNAL_update_send(void *userData, uint32_t addr, char *data)
} }
cd->flags = cd->flags & 0xFFFFFFFE; cd->flags = cd->flags & 0xFFFFFFFE;
if(cd->sendPktQueue->size == 0 && cd->resendPktQueue->size == 0)
if((cd->flags & 0x8) != 0)
{
// initiate connection to server
if(UDPC_INTERNAL_ts_diff(&us->tsNow, &cd->sent) < UDPC_INIT_PKT_INTERVAL)
{
return;
}
cd->sent = us->tsNow;
char *data = malloc(20);
UDPC_INTERNAL_prepare_pkt(
data,
UDPC_ID_CONNECT,
0,
0xFFFFFFFF,
NULL,
0);
struct sockaddr_in destinationInfo;
destinationInfo.sin_family = AF_INET;
destinationInfo.sin_addr.s_addr = addr;
destinationInfo.sin_port = htons(cd->port);
long int sentBytes = sendto(
us->ctx->socketHandle,
data,
20,
0,
(struct sockaddr*) &destinationInfo,
sizeof(struct sockaddr_in));
if(sentBytes != 20)
{
UDPC_INTERNAL_log(us->ctx, 0, "Failed to send init packet to %s "
"port %d", UDPC_INTERNAL_atostr(us->ctx, addr), cd->port);
free(data);
return;
}
return;
}
if(cd->sendPktQueue->size == 0 && cd->priorityPktQueue->size == 0)
{ {
// send and resend packet queue is empty, send heartbeat packet // send and resend packet queue is empty, send heartbeat packet
if(UDPC_INTERNAL_ts_diff(&us->tsNow, &cd->sent) < UDPC_HEARTBEAT_PKT_INTERVAL) if(UDPC_INTERNAL_ts_diff(&us->tsNow, &cd->sent) < UDPC_HEARTBEAT_PKT_INTERVAL)
@ -775,15 +858,15 @@ void UDPC_INTERNAL_update_send(void *userData, uint32_t addr, char *data)
} }
free(data); free(data);
} }
else // sendPktQueue or resendPktQueue not empty else // sendPktQueue or priorityPktQueue not empty
{ {
UDPC_INTERNAL_PacketInfo *pinfo; UDPC_INTERNAL_PacketInfo *pinfo;
int isResendingPkt = 0; int isResendingPkt = 0;
if(cd->resendPktQueue->size != 0) if(cd->priorityPktQueue->size != 0)
{ {
pinfo = UDPC_Deque_get_front_ptr( pinfo = UDPC_Deque_get_front_ptr(
cd->resendPktQueue, sizeof(UDPC_INTERNAL_PacketInfo)); cd->priorityPktQueue, sizeof(UDPC_INTERNAL_PacketInfo));
isResendingPkt = 1; isResendingPkt = 1;
} }
else else
@ -820,7 +903,7 @@ void UDPC_INTERNAL_update_send(void *userData, uint32_t addr, char *data)
free(pinfo->data); free(pinfo->data);
if(isResendingPkt != 0) if(isResendingPkt != 0)
{ {
UDPC_Deque_pop_front(cd->resendPktQueue, sizeof(UDPC_INTERNAL_PacketInfo)); UDPC_Deque_pop_front(cd->priorityPktQueue, sizeof(UDPC_INTERNAL_PacketInfo));
} }
else else
{ {
@ -871,7 +954,7 @@ void UDPC_INTERNAL_update_send(void *userData, uint32_t addr, char *data)
free(pinfo->data); free(pinfo->data);
if(isResendingPkt != 0) if(isResendingPkt != 0)
{ {
UDPC_Deque_pop_front(cd->resendPktQueue, sizeof(UDPC_INTERNAL_PacketInfo)); UDPC_Deque_pop_front(cd->priorityPktQueue, sizeof(UDPC_INTERNAL_PacketInfo));
} }
else else
{ {
@ -973,7 +1056,7 @@ void UDPC_INTERNAL_check_pkt_timeout(
pinfo->data = NULL; pinfo->data = NULL;
pinfo->size = 0; pinfo->size = 0;
UDPC_Deque_push_back( UDPC_Deque_push_back(
cd->resendPktQueue, cd->priorityPktQueue,
&newPkt, &newPkt,
sizeof(UDPC_INTERNAL_PacketInfo)); sizeof(UDPC_INTERNAL_PacketInfo));
} }
@ -1091,8 +1174,17 @@ void UDPC_INTERNAL_prepare_pkt(
temp = htonl(conID | ((flags & 0x2) != 0 ? UDPC_ID_RESENDING : 0)); temp = htonl(conID | ((flags & 0x2) != 0 ? UDPC_ID_RESENDING : 0));
memcpy(&d[4], &temp, 4); memcpy(&d[4], &temp, 4);
} }
if(seqID)
{
temp = htonl(*seqID); temp = htonl(*seqID);
++(*seqID); ++(*seqID);
}
else
{
temp = 0;
}
memcpy(&d[8], &temp, 4); memcpy(&d[8], &temp, 4);
temp = htonl(rseq); temp = htonl(rseq);
memcpy(&d[12], &temp, 4); memcpy(&d[12], &temp, 4);

View file

@ -56,6 +56,7 @@ typedef struct {
* 0x1 - is resending * 0x1 - is resending
* 0x2 - is check received packet * 0x2 - is check received packet
* 0x4 - has been re-sent * 0x4 - has been re-sent
* 0x8 - initiate connection packet
*/ */
uint32_t flags; uint32_t flags;
char *data; // no-header in sendPktQueue and receivedPackets, header in sentPkts char *data; // no-header in sendPktQueue and receivedPackets, header in sentPkts
@ -69,6 +70,7 @@ typedef struct {
* 0x1 - trigger send * 0x1 - trigger send
* 0x2 - is good mode * 0x2 - is good mode
* 0x4 - is good rtt * 0x4 - is good rtt
* 0x8 - is id not set yet / initiating connection to server
*/ */
uint32_t flags; uint32_t flags;
uint32_t id; uint32_t id;
@ -83,7 +85,7 @@ typedef struct {
uint16_t port; uint16_t port;
UDPC_Deque *sentPkts; UDPC_Deque *sentPkts;
UDPC_Deque *sendPktQueue; UDPC_Deque *sendPktQueue;
UDPC_Deque *resendPktQueue; UDPC_Deque *priorityPktQueue;
struct timespec received; struct timespec received;
struct timespec sent; struct timespec sent;
float rtt; float rtt;
@ -154,6 +156,8 @@ void UDPC_set_callback_received(
void UDPC_check_events(UDPC_Context *ctx); void UDPC_check_events(UDPC_Context *ctx);
void UDPC_client_initiate_connection(UDPC_Context *ctx, uint32_t addr, uint16_t port);
/*! /*!
* \brief Queues a packet to send to a connected peer * \brief Queues a packet to send to a connected peer
* Note addr is expected to be in network-byte-order (big-endian). * Note addr is expected to be in network-byte-order (big-endian).