-
Notifications
You must be signed in to change notification settings - Fork 6
/
atp_socket.cpp
2204 lines (2085 loc) · 85 KB
/
atp_socket.cpp
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
* Calvin Neo
* Copyright (C) 2017 Calvin Neo <calvinneo@calvinneo.com>
* https://github.com/CalvinNeo/ATP
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 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, write to the Free Software Foundation, Inc.,
* 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
*/
#include "atp_impl.h"
#include "udp_util.h"
#include <algorithm>
#include <bitset>
#include <iostream>
ATPSocket::ATPSocket(ATPContext * _context) : context(_context){
assert(context != nullptr);
sock_id = context->new_sock_id();
conn_state = CS_UNINITIALIZED;
memset(hash_str, 0, sizeof hash_str);
memset(callbacks, 0, sizeof callbacks);
init_callbacks(this);
}
atp_callback_arguments ATPSocket::make_atp_callback_arguments(ATP_CALLBACKTYPE_ENUM method, OutgoingPacket * out_pkt, const ATPAddrHandle & addr){
if(out_pkt == nullptr){
// there's no OutgoingPacket to be sent, so pass `nullptr`
// out_pkt->length = 0, out_pkt->data = nullptr
return atp_callback_arguments{
context,
this,
method,
0, nullptr,
conn_state,
reinterpret_cast<const SA*>(&(addr.sa)),
sizeof(sockaddr_in)
};
}else{
return atp_callback_arguments{
context,
this,
method,
out_pkt->length, out_pkt->data,
conn_state,
reinterpret_cast<const SA*>(&(addr.sa)),
sizeof(sockaddr_in)
};
}
}
OutgoingPacket * ATPSocket::basic_send_packet(uint16_t flags){
// use `{{}}` to make C++14 happy
ATPPacket pkt = ATPPacket{
(uint16_t)(seq_nr & seq_nr_mask), // seq_nr, updated in send_packet
(uint16_t)(ack_nr & seq_nr_mask), // ack_nr
(uint16_t)peer_sock_id, // peer_sock_id
0,// opts_count
(uint8_t)flags, // flags
(uint16_t)my_window // my window
};
OutgoingPacket * out_pkt = new OutgoingPacket{
0, // observer
0, // marked
0, // selective_acked
0, // ahead_handled
0, // need_resend, update by `send_packet`/`check_unsend_packet`
sizeof (ATPPacket), // length, update by `add_data`
0, // payload, update by `add_data`/`add_option`
0, // option_len, update by `add_option`
0, // timestamp, set at `send_packet_noguard`
0, // transmissions, update by `send_packet_noguard`
seq_nr, // full_seq_nr, updated in send_packet
reinterpret_cast<char *>(std::calloc(1, sizeof (ATPPacket))) // SYN packet will not contain data
};
std::memcpy(out_pkt->data, &pkt, sizeof (ATPPacket));
return out_pkt;
}
void ATPSocket::clear(){
for(OutgoingPacket * op : outbuf){
delete op;
}
outbuf.clear();
for(OutgoingPacket * op : inbuf){
delete op;
}
inbuf.clear();
}
#ifdef _ATP_CLEAR_STATE_FUNC
void ATPSocket::clear_state(){
sock_id = context->new_sock_id();
memset(hash_str, 0, sizeof hash_str);
memset(callbacks, 0, sizeof callbacks);
init_callbacks(this);
peer_sock_id = 0;
dest_addr = ATPAddrHandle();
seq_nr = 0;
ack_nr = 0;
overflow_lock = false;
new_stage_hitted = false;
peer_seq_nr_base = 0;
my_seq_acked_by_peer = 0;
rtt = 0;
rtt_var = 800;
rto = 2000;
ack_delayed_time = 200;
delay_ack_timeout = 0;
rto_timeout = 0;
death_timeout = 0;
persist_timeout = 0;
transmission_counter = 0;
atp_retries1 = 3;
atp_retries2 = 8;
atp_syn_retries1 = 5;
atp_frr_retries = 0;
frr_counter = 0;
cur_window_packets = window_packets_unlimited;
used_window_packets = 0;
enable_cork = false;
cur_window = window_packets_unlimited;
used_window = 0;
peer_window = window_packets_unlimited;
my_window = window_packets_unlimited;
current_mss = ATP_MSS_CEILING;
reorder_count = 0;
#ifdef USE_OLD_SACK_FIELD
peer_max_sack_count = 0;
my_max_sack_count = 5;
#else
peer_max_sack_count = 0;
my_max_sack_count = 4;
#endif
// Options
reuse_port_flag = false;
on_listen_port = false;
last_receive_timestamp = 0;
last_send_timestamp = 0;
}
#endif
int ATPSocket::init(int family, int type, int protocol){
switch_state(CS_IDLE);
sockfd = socket(family, type, protocol);
get_local_addr().family() = family;
dest_addr.family() = family;
#if defined (ATP_LOG_AT_DEBUG) && defined(ATP_LOG_UDP)
log_debug(this, "UDP Socket init, sockfd %d.", sockfd);
#endif
return sockfd;
}
int ATPSocket::init_fork(ATPSocket * origin){
switch_state(CS_IDLE);
sockfd = origin->sockfd;
family = origin->family;
type = origin->type;
protocol = origin->protocol;
get_local_addr().family() = family;
dest_addr.family() = family;
reuse_port_flag = true;
#if defined (ATP_LOG_AT_DEBUG) && defined(ATP_LOG_UDP)
log_debug(this, "UDP Socket forked from %s, sockfd %d.", origin->to_string(), sockfd);
#endif
return sockfd;
}
ATP_PROC_RESULT ATPSocket::connect(const ATPAddrHandle & to_addr){
assert(context != nullptr);
dest_addr = to_addr;
assert(conn_state == CS_IDLE);
context->register_to_look_up(this);
while(seq_nr == 0){
#if defined (ATP_DEBUG_TEST_OVERFLOW)
seq_nr = 0xfffd;
#else
seq_nr = rand() & seq_nr_mask;
#endif
}
ack_nr = 0;
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_SYN));
add_option(out_pkt, ATP_OPT_SOCKID, sizeof(sock_id), reinterpret_cast<char*>(&sock_id));
if (my_max_sack_count > 0)
{
add_option(out_pkt, ATP_OPT_SACKOPT, sizeof(my_max_sack_count), reinterpret_cast<char*>(&my_max_sack_count));
}
// before sending packet, users can do something, like call `connect` to their UDP socket.
atp_callback_arguments arg = make_atp_callback_arguments(ATP_CALL_CONNECT, out_pkt, dest_addr);
ATP_PROC_RESULT result = invoke_callback(ATP_CALL_CONNECT, &arg);
#if defined (ATP_LOG_AT_DEBUG) && defined(ATP_LOG_UDP)
log_debug(this, "UDP socket connect to %s.", dest_addr.to_string());
#endif
if (result == ATP_PROC_ERROR){
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Connect Failed.");
#endif
delete out_pkt;
out_pkt = nullptr;
} else{
result = send_packet(out_pkt);
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Sent SYN to peer, seq:%u.", out_pkt->get_head()->seq_nr);
#endif
}
switch_state(CS_SYN_SENT);
return result;
}
ATP_PROC_RESULT ATPSocket::listen(uint16_t host_port){
switch_state(CS_LISTEN);
// set port
get_local_addr().set_port(host_port);
// register this port to context's listen list
if (context->register_listen_port(this, host_port) == ATP_PROC_OK)
{
this->on_listen_port = true;
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "I am listening on port %u.", host_port);
#endif
}
}
ATP_PROC_RESULT ATPSocket::bind(const ATPAddrHandle & to_addr){
// There's no OutgoingPacket to be sent, so pass `nullptr`
atp_callback_arguments arg = make_atp_callback_arguments(ATP_CALL_BIND, nullptr, to_addr);
ATP_PROC_RESULT result = invoke_callback(ATP_CALL_BIND, &arg);
return result;
}
ATPSocket * ATPSocket::fork_basic(){
// Forked socket ans origin socket share the same sockfd,
// and is distinguished by sock_id
ATPSocket * socket = new ATPSocket(context);
int sockfd = socket->init_fork(this);
context->sockets.push_back(socket);
return socket;
}
ATPSocket * ATPSocket::fork_me(){
return fork_basic();
}
ATP_PROC_RESULT ATPSocket::accept(const ATPAddrHandle & to_addr, OutgoingPacket * recv_pkt){
ATP_PROC_RESULT result;
assert(context != nullptr);
if (conn_state == CS_LISTEN || conn_state == CS_IDLE)
{
// This is the first SYN request
// if(re_listen) {
// goto REUSED_SOCKET;
// }
}else if(reuse_port_flag) {
// Port reusing is allowed, so this socket will not be removed from listening queue.
REUSED_SOCKET:
re_listen = false;
char * _opt_sockid_field = recv_pkt->find_option(ATP_OPT_SOCKID);
if(_opt_sockid_field == nullptr) return ATP_PROC_OK; // This is a bad packet. Ignore it!
uint16_t _peer_sock_id = *reinterpret_cast<uint16_t*>(_opt_sockid_field + 2 * sizeof(uint8_t));
if (this->peer_sock_id == _peer_sock_id)
{
// This packet maybe a SYN packet from peer but for some reason delayed
// Just ignore, don't need ACK, because my SYN-ACK will automaticlly re-send
return ATP_PROC_OK;
}else{
// In some implementation of TCP protocol, new connection can establish at TIME_WAIT state.
// But ATP don't need that because ATP can have very short TIME_WAIT time.
// We treat it as a new connection to another socket listening at the same port.
atp_callback_arguments arg = make_atp_callback_arguments(ATP_CALL_BEFORE_REP_ACCEPT, nullptr, to_addr);
result = invoke_callback(ATP_CALL_BEFORE_REP_ACCEPT, &arg);
if (result == ATP_PROC_REJECT)
{
return ATP_PROC_OK;
}else {
// atp_callback_arguments::socket is `newsock` rather than `this`
arg = make_atp_callback_arguments(ATP_CALL_ON_FORK, nullptr, to_addr);
arg.socket = this;
// After this call, arg.socket will be set to a newly created socket.
result = invoke_callback(ATP_CALL_ON_FORK, &arg);
if (result != ATP_PROC_OK || arg.socket == this)
{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "You must create a new socket.");
#endif
return ATP_PROC_ERROR;
}else{
ATPSocket * newsock = arg.socket;
newsock->accept(to_addr, recv_pkt);
return ATP_PROC_OK;
}
}
}
} else {
// In this case, corresponding packets won't be dispatched to socket.
return ATP_PROC_OK;
}
atp_callback_arguments arg = make_atp_callback_arguments(ATP_CALL_BEFORE_ACCEPT, nullptr, to_addr);
result = invoke_callback(ATP_CALL_BEFORE_ACCEPT, &arg);
#if defined (ATP_LOG_AT_NOTE)
print_out(this, recv_pkt, "rcv");
#endif
if (result == ATP_PROC_OK)
{
// accept this SYN
dest_addr = to_addr;
// Now our socket can be located
if(!reuse_port_flag){
// If we don't need to reuse port, we will remove socket from listening
on_listen_port = false;
context->deregister_listen_port(get_local_addr().host_port());
}
context->register_to_look_up(this);
#if defined (ATP_LOG_AT_NOTE)
print_out(this, recv_pkt, "rcv");
#endif
init_connection(recv_pkt, false);
}else if(result == ATP_PROC_REJECT){
// decline this SYN
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_RST));
result = send_packet(out_pkt);
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Reject SYN request from %s by sending RST.", ATPSocket::make_hash_code(peer_sock_id, dest_addr));
#endif
}else{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "ERROR.");
#endif
}
return result;
}
ATP_PROC_RESULT ATPSocket::receive(OutgoingPacket * recv_pkt, size_t real_payload_offset){
if (recv_pkt->get_head()->get_fin())
{
// cond2: ignore fin
// just ignore
return ATP_PROC_OK;
}else if(recv_pkt->get_head()->get_syn()){
// the 2 bytes payload in syn packet are not user data, they carried sock_id
return ATP_PROC_OK;
}else if(!recv_pkt->has_user_data()){
// there is no payload
// just ignore
return ATP_PROC_OK;
}else{
atp_callback_arguments arg = make_atp_callback_arguments(ATP_CALL_ON_RECV, recv_pkt, dest_addr);
arg.data = recv_pkt->data + sizeof(ATPPacket) + real_payload_offset;
arg.length = recv_pkt->payload - real_payload_offset;
size_t actual_received = invoke_callback(ATP_CALL_ON_RECV, &arg);
// TODO FIXME IMPORTANT handle situation when this is not drained read
if (actual_received < arg.length)
{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "User read only %u of %u bytes.", actual_received, arg.length);
#endif
}else{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "User read %u bytes(drained).", actual_received);
#endif
return ATP_PROC_OK;
}
}
}
ATP_PROC_RESULT ATPSocket::send_packet_noguard(OutgoingPacket * out_pkt, bool adhoc){
// argument `adhoc`, which is usually set to false, is used by debuggers who can send simulated packets by `send_packet_noguard`
uint64_t current_ms = get_current_ms();
rto_timeout = current_ms + rto;
if (out_pkt->transmissions == 0 && out_pkt->is_promised_packet() && !out_pkt->get_head()->get_urg() && !adhoc)
{
used_window_packets++;
used_window += out_pkt->payload;
}
#if defined (ATP_LOG_AT_DEBUG)
if(out_pkt->get_head()->get_urg()){
log_debug(this, "ATPPacket URG sent. seq:%u size:%u payload:%u.", out_pkt->get_head()->seq_nr, out_pkt->length, out_pkt->payload);
}else{
log_debug(this, "ATPPacket sent. seq:%u size:%u payload:%u.", out_pkt->get_head()->seq_nr, out_pkt->length, out_pkt->payload);
}
#endif
#if (defined (ATP_LOG_AT_NOTE)) || (defined (ATP_LOG_AT_DEBUG))
char b[32];
if (out_pkt->transmissions == 0)
{
if(out_pkt->get_head()->get_urg()){
sprintf(b, "urg");
}else{
sprintf(b, "snd");
}
}else{
sprintf(b, "snd[%u]", out_pkt->transmissions);
}
#ifdef ATP_LOG_AT_NOTE
print_out(this, out_pkt, b);
#endif
#ifdef ATP_LOG_AT_DEBUG
print_out(this, out_pkt, b, stderr);
#endif
#endif
out_pkt->timestamp = current_ms;
out_pkt->transmissions++;
atp_callback_arguments arg = make_atp_callback_arguments(ATP_CALL_SENDTO, out_pkt, dest_addr);
if (out_pkt->need_resend)
{
out_pkt->need_resend = false;
}
ATP_PROC_RESULT result = invoke_callback(ATP_CALL_SENDTO, &arg);
return result;
}
void ATPSocket::check_unsend_packet(){
// This function send all packets which is allowed to be sent, and haven't yet been sent, including
// 1. New packets from outbuf. These packets have `transmissions == 0`
// 2. Packets need re-sending, due to timeout. These packets have `need_resend == true`
if(outbuf.size() <= 0) {return;} // If there's no cached packets
int marked_total = 0;
for(OutgoingPacket * out_pkt : outbuf){
// Check everytime in the for-loop
if (out_pkt && (out_pkt->transmissions == 0 || out_pkt->need_resend))
{
// For most cases, need_resend will not be true, because currently delayed ACK checks before timeout check.
if (out_pkt->is_empty_ack())
{
#if defined (ATP_LOG_AT_DEBUG)
// In previoud version if delayed ack is enabled, an empty ACK packet is allowed to be sent later
// We don't send an empty packet with old ack_nr.
// This can't happen now. Here also deleted some codes which mark all empty packets and then delete them
bool cant_happen = guess_full_seq_nr(out_pkt->get_head()->ack_nr) < ack_nr && !overflow_lock;
assert(!cant_happen);
#endif
send_packet_noguard(out_pkt);
}else if(out_pkt->transmissions > 0){
send_packet_noguard(out_pkt);
}else if(!is_full(out_pkt->payload)){
send_packet_noguard(out_pkt);
}
}
}
delay_ack_timeout = 0;
}
ATP_PROC_RESULT ATPSocket::send_packet(OutgoingPacket * out_pkt, bool flush_packets, bool adhoc){
ATP_PROC_RESULT result = ATP_PROC_OK;
// Setup packets
// When the package is constructed, update `seq_nr` for the next package
if (out_pkt->is_promised_packet() && !out_pkt->need_resend)
{
// Only non-repeated packets with real user data(not including options) can have new seq number
out_pkt->full_seq_nr ++;
out_pkt->get_head()->seq_nr ++;
(out_pkt->get_head()->seq_nr) &= seq_nr_mask;
seq_nr ++;
}
if(adhoc){
// Argument `adhoc` functions the same as which in `send_packet_noguard`
result = send_packet_noguard(out_pkt, true);
return result;
}
else if(out_pkt->get_head()->get_urg()){
// URG packets are always sent immediately
result = send_packet_noguard(out_pkt);
PUSH_OUTBUF(out_pkt);
}
else if (bytes_can_send_one_packet() >= out_pkt->payload) {
// Actually send
// If payload == 0 and is not FIN, can always send regradless of window
if (out_pkt->is_empty_ack())
{
// Send Empty ACK immediately
// ACK packet is ad-hoc sent packet, don't enqueue to `outbuf`(always delete at once(except SYN, FIN))
result = send_packet_noguard(out_pkt);
delete out_pkt;
out_pkt = nullptr;
// Cancel schedule_ack
delay_ack_timeout = 0;
}else if(out_pkt->get_head()->get_syn()){
// SYN Packet will always be sent immediately and pushed into `outbuf`
result = send_packet_noguard(out_pkt);
PUSH_OUTBUF(out_pkt);
}else{
// We don't send immediately packets with userdata.
// Enqueue `out_pkt` for proper time to send and for potential resend
PUSH_OUTBUF(out_pkt);
if(flush_packets){
check_unsend_packet();
}
}
}else{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "ATPPacket sent CACHED due to window limit. seq:%u size:%u payload:%u."
, out_pkt->get_head()->seq_nr, out_pkt->length, out_pkt->payload);
#endif
PUSH_OUTBUF(out_pkt);
}
if (out_pkt != nullptr)
{
// Cancel schedule_ack
assert(out_pkt->data != nullptr);
if(out_pkt->get_head()->get_ack()){
delay_ack_timeout = 0;
}
}
return result;
}
ATP_PROC_RESULT ATPSocket::close(){
int result = ATP_PROC_OK;
switch(conn_state){
case CS_UNINITIALIZED:
case CS_IDLE:
case CS_SYN_SENT:
case CS_SYN_RECV:
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Close: Close a socket which is not connected yet.");
#endif
result = ATP_PROC_ERROR;
break;
case CS_LISTEN:
case CS_PASSIVE_LISTEN:
{
// Call `close` on a listening socket is valid operation
switch_state(CS_DESTROY);
result = ATP_PROC_OK;
break;
}
case CS_CONNECTED:
case CS_CONNECTED_FULL:
{
// A
switch_state(CS_FIN_WAIT_1);
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Close: Send FIN Packet.");
#endif
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_FIN));
// FIN must have payload, and it's own seq_nr, otherwise, It will share `seq_nr` data with a data packet,
// supposing this data packet is received normally, and FIN lost.
// I can get a wrong ACK for data packet from peer, and mistaken it as an ack for my FIN packet
// In this case, we will never re-send our FIN packet, so peer can never know we finished.
add_data(out_pkt, &sock_id, sizeof(sock_id));
send_packet(out_pkt);
result = ATP_PROC_OK;
break;
}
case CS_FIN_WAIT_1:
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Close: Already sent FIN, waiting for ACK.");
#endif
result = ATP_PROC_ERROR;
break;
case CS_CLOSE_WAIT:
{
// B
switch_state(CS_LAST_ACK);
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Close: Send FIN Packet to a finished peer.");
#endif
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_FIN));
add_data(out_pkt, &sock_id, sizeof(sock_id));
send_packet(out_pkt);
result = ATP_PROC_OK;
break;
}
case CS_FIN_WAIT_2:
case CS_LAST_ACK:
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Close: Connection Error.");
#endif
result = ATP_PROC_ERROR;
break;
case CS_TIME_WAIT:
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Close: Send a repeated last ACK.");
#endif
result = ATP_PROC_OK;
break;
case CS_RESET:
case CS_DESTROY:
default:
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Close: Connection Error.");
#endif
result = ATP_PROC_ERROR;
break;
}
return result;
}
void ATPSocket::add_data(OutgoingPacket * out_pkt, const void * buf, const size_t len){
(out_pkt->length) += len;
(out_pkt->payload) += len;
assert(out_pkt->length == out_pkt->payload + sizeof(ATPPacket));
out_pkt->data = reinterpret_cast<char *>(std::realloc(out_pkt->data, out_pkt->length));
memcpy(out_pkt->data + (out_pkt->length - len), buf, len);
assert(out_pkt->data != nullptr);
}
void ATPSocket::add_option(OutgoingPacket * out_pkt, uint8_t opt_kind, uint8_t opt_data_len, char * opt_data){
out_pkt->get_head()->opts_count++;
uint8_t opt_len = opt_data_len + sizeof(uint8_t) * 2;
size_t prev_opt_len = out_pkt->option_len;
out_pkt->data = reinterpret_cast<char *>(std::realloc(out_pkt->data, out_pkt->length + opt_len));
if (out_pkt->real_payload() != 0)
{
// If there is user data, must move them to make room for new options
std::memmove(out_pkt->data + sizeof(ATPPacket) + prev_opt_len + opt_len, out_pkt->data + sizeof(ATPPacket) + prev_opt_len, out_pkt->real_payload());
}
*reinterpret_cast<uint8_t *>(out_pkt->data + out_pkt->length) = opt_kind;
*reinterpret_cast<uint8_t *>(out_pkt->data + out_pkt->length + sizeof(uint8_t)) = opt_data_len;
memcpy(out_pkt->data + out_pkt->length + sizeof(uint8_t) * 2, opt_data, opt_data_len);
out_pkt->length += opt_len;
out_pkt->payload += opt_len;
out_pkt->option_len += opt_len;
assert(out_pkt->length == out_pkt->payload + sizeof(ATPPacket));
#if defined (ATP_LOG_AT_DEBUG)
// fprintf(stderr, "Add option[%u], kind:%u, data_len:%u, payload:%u\n", out_pkt->get_head()->opts_count
// , opt_kind, opt_data_len, out_pkt->payload);
#endif
}
bool ATPSocket::writable() const{
// NOTICE: note that even window is 0, this socket is still writable
// A writable socket can accept inputs
switch(conn_state){
case CS_UNINITIALIZED:
case CS_IDLE:
case CS_LISTEN:
case CS_PASSIVE_LISTEN:
case CS_SYN_SENT:
case CS_SYN_RECV:
case CS_RESET:
// can't write data now
return false;
case CS_CONNECTED:
case CS_CONNECTED_FULL:
return true;
case CS_FIN_WAIT_1: // A
// I closed, I can't write, but peer can
return false;
case CS_CLOSE_WAIT: // B
// I closed, I can't write, but peer can
return true;
case CS_FIN_WAIT_2: // A
// I already closed
return false;
case CS_LAST_ACK: // B
// I finished, peer finished
return false;
case CS_TIME_WAIT:
case CS_DESTROY:
default:
return false;
}
}
bool ATPSocket::readable() const{
// A readable socket is capable of reading data from peer
switch(conn_state){
case CS_UNINITIALIZED:
case CS_IDLE:
case CS_LISTEN:
case CS_SYN_SENT:
case CS_SYN_RECV:
case CS_RESET:
// Now connection not yet been established
return false;
case CS_CONNECTED:
case CS_CONNECTED_FULL: // B
return true;
case CS_FIN_WAIT_1: // A
// Can still read
return true;
case CS_CLOSE_WAIT: // B
// Peer closed, can't read
return false;
case CS_FIN_WAIT_2: // A
// Peer not closed, can still read
return true;
case CS_LAST_ACK: // B
return false;
case CS_TIME_WAIT: // A
return false;
case CS_DESTROY:
case CS_PASSIVE_LISTEN:
default:
return false;
}
}
bool ATPSocket::eof() const{
// When a socket is not readable after connection is established, eof = true
switch(conn_state){
case CS_UNINITIALIZED:
case CS_IDLE:
case CS_LISTEN:
case CS_SYN_SENT:
case CS_SYN_RECV:
case CS_RESET:
case CS_CONNECTED:
case CS_CONNECTED_FULL: // B
case CS_FIN_WAIT_1: // A
// Can still read
return false;
case CS_CLOSE_WAIT: // B
// Peer closed, can't read
return true;
case CS_FIN_WAIT_2: // A
// Peer not closed, can still read
return false;
case CS_LAST_ACK: // B
case CS_TIME_WAIT: // A
case CS_PASSIVE_LISTEN:
case CS_DESTROY:
return true;
default:
return true;
}
}
size_t ATPSocket::bytes_can_send_once() const {
if(cur_window == window_packets_unlimited){
// Notice cur_window is now uint16_t and we are supposed to return a size_t
// So we keep window_packets_unlimited to be int
return window_packets_unlimited;
}
else if (cur_window < used_window)
{
// If window is already used up
return 0;
}
return std::min(cur_window - used_window, MAX_ATP_PAYLOAD);
}
size_t ATPSocket::bytes_can_send_one_packet(OutgoingPacket * particular_packet) const {
if(particular_packet == nullptr){
return std::min(bytes_can_send_once(), current_mss);
}else{
return current_mss - particular_packet->payload;
}
}
ATP_PROC_RESULT ATPSocket::write_oob(const void * buf, const size_t len, uint32_t timeout){
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_URG));
add_data(out_pkt, buf, len);
ATP_PROC_RESULT result = send_packet(out_pkt);
return result;
}
ATP_PROC_RESULT ATPSocket::write(const void * buf, const size_t len){
if (!writable())
{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "ERROR: This socket is not writable.");
#endif
return ATP_PROC_ERROR;
}
// TODO improve here
#if defined (ATP_LOG_AT_DEBUG)
if (len > bytes_can_send_one_packet())
{
// If mss restriction not satisfied
log_debug(this, "Must devide into several ATP packets.");
}
#endif
size_t p = 0; int packet_id = 0;
while(p < len){
if (bytes_can_send_once() == 0)
{
// If window restriction not satisfied
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Window restricted, total %u, used %u, will only partially send, cache the rest.", cur_window, used_window);
#endif
break;
}
// TODO reuse packets
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_ACK));
size_t add_len = fill_packet(out_pkt, buf + p, len - p);
// Don't immediately emit packet here, flush them together.
ATP_PROC_RESULT result = send_packet(out_pkt, false);
if (result == ATP_PROC_ERROR)
{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "In packet_id:%d, write %u bytes to peer from position %u, seq:%u."
, packet_id, out_pkt->payload, p, seq_nr);
#endif
return ATP_PROC_ERROR;
}else{
packet_id++;
p += add_len;
}
}
// Flush packets when all packets are created
check_unsend_packet();
// Successfully sent `p` bytes of data
return p;
}
ATP_PROC_RESULT ATPSocket::check_fin(OutgoingPacket * recv_pkt){
// return >0: OK
// return -1: error
uint64_t current_ms = get_current_ms();
ATP_PROC_RESULT result = ATP_PROC_OK;
switch(conn_state){
case CS_UNINITIALIZED:
case CS_IDLE:
case CS_SYN_SENT:
case CS_SYN_RECV:
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Connection Error.");
#endif
result = ATP_PROC_ERROR;
break;
case CS_CONNECTED:
case CS_CONNECTED_FULL: // B
{
switch_state(CS_CLOSE_WAIT);
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Recv peer's FIN, Send the last ACK to Peer, Me still alive.");
#endif
schedule_ack();
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Call ATP_CALL_ON_PEERCLOSE.");
#endif
atp_callback_arguments arg = make_atp_callback_arguments(ATP_CALL_ON_PEERCLOSE, recv_pkt, dest_addr);
result = invoke_callback(ATP_CALL_ON_PEERCLOSE, &arg);
// half closed, don't send FIN immediately
break;
}
case CS_FIN_WAIT_1: // A
case CS_CLOSE_WAIT: // B
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Connection Error.");
#endif
result = ATP_PROC_ERROR;
break;
case CS_FIN_WAIT_2: // A
{
// Receive FIN from peer
switch_state(CS_TIME_WAIT);
uint64_t current_ms = get_current_ms();
death_timeout = current_ms + std::max(context->min_msl2, rto);
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Recv peer's FIN, Send the last ACK to Peer, wait 2MSL from %u to %u.", current_ms, death_timeout);
#endif
// Send no scheduled ACK
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_ACK));
result = send_packet(out_pkt);
atp_callback_arguments arg;
// arg = make_atp_callback_arguments(ATP_CALL_ON_PEERCLOSE, recv_pkt, dest_addr);
// result = context->callbacks[ATP_CALL_ON_PEERCLOSE](&arg);
break;
}
case CS_LAST_ACK: // B
{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Peer resend FIN, send ack.");
#endif
// Send no scheduled ACK
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_ACK));
result = send_packet(out_pkt);
break;
}
case CS_TIME_WAIT: // A
{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Peer resend FIN, send ack.");
#endif
death_timeout = current_ms + std::max(context->min_msl2, rto);
// Send no scheduled ACK
OutgoingPacket * out_pkt = basic_send_packet(ATPPacket::create_flags(PACKETFLAG_ACK));
result = send_packet(out_pkt);
break;
}
case CS_RESET:
case CS_DESTROY:
default:
{
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Connection Error.");
#endif
result = ATP_PROC_ERROR;
break;
}
}
return result;
}
uint32_t ATPSocket::guess_full_seq_nr(uint32_t raw_peer_seq){
// There's a method to determine how many packets are our received packet ahead/behind current ack_nr employing a overflow strategy
// e.g. when received sequence number is 1 and current ack_nr is 0xff,
// `1 - 0xff = 2` means the packet is 2 packets ahead of current ack.
// However consider a old packet 0xfe, `0xfe - 0xff = 255` means the packet is 255 packets ahead of(in fact 1 packet behind) the current ack
// Though we can't tell the exact difference between ahead or behind.
// The current method can judge whether a packet is ahead/behind, because we have base and full length ack_nr/seq_nr/peer_seq_nr
uint32_t peer_seq = get_full_seq_nr(raw_peer_seq);
// But if then we get a packet who somehow comes later, with an un-overflowed seq_nr?
// e.g. If seq_nr_mask is 0xff(255), and our ack_nr is now 1025(0x3ff + 2).
// If we then get a packet of 2, that's obviously an expected packet, and we can update ack_nr to 1026
// What if we get a packet of 254? how can we judge whether it means 1023(0x3ff) or 1274(0x4fa)=1024+254?
// To solve this, we have a 2 assumptions:
// 1. when handling packet numbered `s`, all packets numbered before `s - (peer_seq_nr_base + 1)` are vanished.
// It means that there is only one packet in the queue with exactly the same seq_nr
// 2. packets are handled generally the same rate
// Now handle this situation, where we got a packet of 254,
// The result depends on relationship between the two possible values and ack_nr
// 1. First of all, we can find out the easiest case, when ack_nr is less than any of the two possible values.
// For example, if ack_nr is now 250 and greater than both 254 and 510, this is possible because packet 256 can come earlier and overflow.
// In this case, the true peer_seq should be 254, because according to our assumption, packet 510 can't co-exist with packet 254
// 2. If ack_nr is between the two possible values.
// For example, if ack_nr is now 257, and we decide between 254 and 510.
// In this case if we choose 254, and then drop it. We can also choose 510, because it's a packet comes much earlier.
// They are two streategies, a) we choose the most close to ack_nr, because we believe in possibility.
// b) We can always choose the smaller, because we can depend on resend.
// 3. If ack_nr is greater than the two possible values.
// For example, if ack_nr is now 555, and we decide between 254 and 510.
// This is also easy, we choose 510.
if (peer_seq > (seq_nr_mask + 1))
{
// notice peer_seq > peer_seq_2 is always true
uint32_t peer_seq_2 = peer_seq - (seq_nr_mask + 1);
if (peer_seq > ack_nr && peer_seq_2 > ack_nr)
{
// not overflow, choose smaller
peer_seq = peer_seq_2;
}else if (peer_seq >= ack_nr && peer_seq_2 <= ack_nr)
{
if (peer_seq - ack_nr >= ack_nr - peer_seq_2)
{
// peer_seq is nearer
peer_seq = peer_seq_2;
}
}else{
// not overflow, choose bigger
}
}
if (peer_seq < ack_nr)
{
// In this case, a re-ordered overflowed packet will be regarded as a very old packet, thus discard
// e.g.
// rcv 504 SO 65534 7 0
// snd 504 SAO 65534 7 65534
// drop 1505 AD 3 1462 65534
// drop 1505 AD 4 1462 65534
uint32_t peer_seq_2 = peer_seq + (seq_nr_mask + 1);
if (ack_nr - peer_seq > peer_seq_2 - ack_nr)
{
peer_seq = peer_seq_2;
}
}
return peer_seq;
}
ATP_PROC_RESULT ATPSocket::update_myack(OutgoingPacket * recv_pkt){
if(conn_state < CS_CONNECTED){
// handles the last hand-shake of connection establishment
#if defined (ATP_LOG_AT_DEBUG)
log_debug(this, "Connection established on B's side, handshake completed.");
#endif
switch_state(CS_CONNECTED);