knetrs 0.1.1

A TCP library
Documentation 
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

#include "example_config.h"

#if COMPILE_MULTI_LOOP

#include "knet.h"

/**
 4�߳�+���߳�
 */

#define MAX_CLIENT 200
#define MAX_ECHO_COUNT MAX_CLIENT * 100
#define MAX_LOOP 4
#define TEST_TIMES 100
int recv_count = 0;
atomic_counter_t client_count = 0;

void client_cb(kchannel_ref_t* channel, knet_channel_cb_event_e e) {
    kaddress_t* peer_address = 0;
    char buffer[10] = {0};
    kstream_t* stream = 0;
    int bytes = 0;
    if (e & channel_cb_event_recv) {
        stream = knet_channel_ref_get_stream(channel);
        bytes = knet_stream_available(stream);
        knet_stream_pop(stream, buffer, sizeof(buffer));
        knet_stream_push(stream, buffer, bytes);
        recv_count++;
        if (recv_count > MAX_ECHO_COUNT) {
            peer_address = knet_channel_ref_get_peer_address(channel);
            printf("client reach max echo, close: %s, %d\n", address_get_ip(peer_address), address_get_port(peer_address));
            knet_channel_ref_close(channel);
        }
    } else if (e & channel_cb_event_timeout) {
        peer_address = knet_channel_ref_get_peer_address(channel);
        printf("client timeout, close: %s, %d\n", address_get_ip(peer_address), address_get_port(peer_address));
        knet_channel_ref_close(channel);
    }
}

int total_connected = 0;

void connector_cb(kchannel_ref_t* channel, knet_channel_cb_event_e e) {
    char buffer[1024] = {0};
    kstream_t* stream = 0;
    int bytes = 0;
    if (e & channel_cb_event_connect) {
        total_connected++;
        printf("connect finished: %d, %d\n", knet_channel_ref_get_socket_fd(channel), total_connected);
    } else if (e & channel_cb_event_recv) {
        stream = knet_channel_ref_get_stream(channel);
        bytes = knet_stream_available(stream);
        bytes = knet_stream_pop(stream, buffer, bytes);
        knet_stream_push(stream, buffer, bytes);
    } else if (e & channel_cb_event_close) {
        atomic_counter_dec(&client_count);
        if (client_count == 0) {
            printf("all client closed\n");
            knet_loop_exit(knet_channel_ref_get_loop(channel));
        }
    } else if (e & channel_cb_event_connect_timeout) {
        printf("connector close, connect timeout: %d\n", knet_channel_ref_get_socket_fd(channel));
        knet_channel_ref_close(channel);
    } else if (e & channel_cb_event_timeout) {
        printf("connector recv timeout, close: %d\n", knet_channel_ref_get_socket_fd(channel));
        knet_channel_ref_close(channel);
    }
}

void acceptor_cb(kchannel_ref_t* channel, knet_channel_cb_event_e e) {
    char buffer[100] = {0};
    kstream_t* stream = 0;
    if (e & channel_cb_event_accept) {
        knet_channel_ref_set_cb(channel, client_cb);
        knet_channel_ref_set_timeout(channel, 2);
        stream = knet_channel_ref_get_stream(channel);
        knet_stream_push(stream, buffer, sizeof(buffer));
    }
}

int main() {
    int i = 0;
    int error = 0;
    kloop_t* main_loop = 0;
    kloop_t* sub_loop[MAX_LOOP] = {0};
    kthread_runner_t* runner[MAX_LOOP] = {0};
    kchannel_ref_t* acceptor = 0;
    kchannel_ref_t* connector = 0;
    kloop_balancer_t* balancer = 0;
    int times = 0;

    /* ����һ�����ؾ����� */
    balancer = knet_loop_balancer_create();
    /* ��������̣߳�ÿ���߳�����һ��kloop_t */
    for (i = 0; i < MAX_LOOP; i++) {
        sub_loop[i] = knet_loop_create();
        knet_loop_balancer_attach(balancer, sub_loop[i]);
        runner[i] = thread_runner_create(0, 0);
        thread_runner_start_loop(runner[i], sub_loop[i], 0);
    }
    main_loop = knet_loop_create();
    knet_loop_balancer_attach(balancer, main_loop);

    acceptor = knet_loop_create_channel(main_loop, 8, 1024 * 8);
    knet_channel_ref_set_cb(acceptor, acceptor_cb);
    error = knet_channel_ref_accept(acceptor, 0, 80, 5000);
    if (error_ok != error) {
        printf("knet_channel_ref_accept failed: %d\n", error);
    }
    
    /* ���� */
    for (; times < TEST_TIMES; times++) {
        total_connected = 0;
        recv_count = 0;
        client_count = MAX_CLIENT;
        for (i = 0; i < MAX_CLIENT; i++) {
            connector = knet_loop_create_channel(main_loop, 8, 8192);
            knet_channel_ref_set_cb(connector, connector_cb);
            knet_channel_ref_set_timeout(connector, 2);
            knet_channel_ref_connect(connector, "127.0.0.1", 80, 2);
        }
        while (client_count > 0) {
            /* ���߳� */
            error = knet_loop_run_once(main_loop);
        }
        if (error != error_ok) {
            printf("knet_loop_run() failed: %d\n", error);
        }
    }
    for (i = 0; i < MAX_LOOP; i++) {
        thread_runner_stop(runner[i]);
        thread_runner_join(runner[i]);
        thread_runner_destroy(runner[i]);
    }
    for (i = 0; i < MAX_LOOP; i++) {
        knet_loop_destroy(sub_loop[i]);
    }
    knet_loop_destroy(main_loop);
    knet_loop_balancer_destroy(balancer);
    return 0;
}

#endif /* COMPILE_MULTI_LOOP */