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
|
/* ==========================================================================
* Licensed under BSD 2clause license See LICENSE file for more information
* Author: Michał Łyszczek <michal.lyszczek@bofc.pl>
* ==========================================================================
* -------------------------------------------------------------
* / This example shows how one can use rb_claim/commit to read \
* | network data in a trivial loop, and then process them in |
* | separate thread in another simple loop. Normally you would |
* | have to deal with mutexes; and memory buffer overlapping |
* | and all of that crap. With librb code can be reduced to |
* | minimum. With rb_claim/commit API you can pass buffer |
* | directly to syscall read() to not perform unnecessary |
* \ copying of data from driver->local_buffer->ring_buffer /
* -------------------------------------------------------------
* \ /\ ___ /\
* \ // \/ \/ \\
* (( O O ))
* \\ / \ //
* \/ | | \/
* | | | |
* | | | |
* | o |
* | | | |
* |m| |m|
* ==========================================================================
* ░▀█▀░█▀█░█▀▀░█░░░█░█░█▀▄░█▀▀░█▀▀
* ░░█░░█░█░█░░░█░░░█░█░█░█░█▀▀░▀▀█
* ░▀▀▀░▀░▀░▀▀▀░▀▀▀░▀▀▀░▀▀░░▀▀▀░▀▀▀
* ========================================================================== */
#include <errno.h>
#include <limits.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
#include "rb.h"
#include "common.h"
/* ==========================================================================
* ░█▀▀░█░█░█▀█░█▀▀░▀█▀░▀█▀░█▀█░█▀█░█▀▀
* ░█▀▀░█░█░█░█░█░░░░█░░░█░░█░█░█░█░▀▀█
* ░▀░░░▀▀▀░▀░▀░▀▀▀░░▀░░▀▀▀░▀▀▀░▀░▀░▀▀▀
* ========================================================================== */
/** =========================================================================
* Function reads data stored in #rb buffer (#arg argument), processes it
* and prints to stdout.
*
* @param arg ring buffer object
* ========================================================================== */
static void *parser(void *arg)
{
for (;;) {
unsigned char d; /* byte of data read from rb */
/* Wait for data to arrive. rb_read() will block thread until data
* arrives */
if (rb_read(arg, &d, sizeof(d)) != 1) {
if (errno == ECANCELED)
/* main thread is going down, not an error, just information
* we should not use rb object anymore, in this case we
* return from the function. This error will happen after
* other thread calls #rb_stop() function. */
return NULL;
/* log, other unexpected error */
perror("rb_read()");
return NULL;
}
/* "parse" data received from network, we just dump received data
* as hex */
if (d == '\n')
printf("\n");
else if (d == '\r')
printf("x%02x(\\r) ", d);
else
printf("x%02x(%c) ", d, d);
}
}
/* ==========================================================================
* ░█▄█░█▀█░▀█▀░█▀█
* ░█░█░█▀█░░█░░█░█
* ░▀░▀░▀░▀░▀▀▀░▀░▀
* ========================================================================== */
int main(void)
{
int cfd; /* server file descriptor */
int sfd; /* connected client file descriptor */
pthread_t parser_t; /* thread that parses messages from network */
struct rb *rb; /* pointer to malloc()ed rb object */
/*~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~*/
/* Create new ring buffer, that can hold 127 elements, each of size 1.
* Make ring buffer multi-thread aware */
if ((rb = rb_new(128, 1, rb_multithread)) == NULL)
pdie("rb_new()");
sfd = start_tcp_server(23893);
printf(
"======================================================================\n"
"server is up, you can send some messages with following commands\n"
"to see program in action (not all might be available on your system\n"
" 1) telnet 127.0.0.1 23893\n"
" 2) echo \"test message\" | nc -w1 127.0.0.1 23893\n"
" 3) echo \"test message\" > /dev/tcp/127.0.0.1/23893\n"
"======================================================================"
"\n\n");
if ((cfd = accept(sfd, NULL, NULL)) < 0)
pdie("accept()");
/* create a thread, that will parse data received from network */
pthread_create(&parser_t, NULL, parser, rb);
/* start a loop, that will read data from network, and will put it
* on ring buffer, so that #parser thread can do something with it */
for (;;) {
ssize_t nread;
void *buffer;
size_t count;
size_t objsize;
/*~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~*/
/* Claim buffer for writing. Reader thread will still be able to read
* data from a queue without being locked out. Function will give us
* location where we can start writing data and how big buffer is */
if (rb_write_claim(rb, &buffer, &count, &objsize))
pdie("rb_write_claim()");
/* Read data from server socket, and place it directly in ring
* buffer's buffer. #buffer is #count * #objsize long, but we know
* that #objsize == 1, so we can just pass #count */
nread = read(cfd, buffer, count);
if (nread == -1)
pdie("read()");
if (nread == 0) {
fprintf(stderr, "Connection closed by remote client\n");
break;
}
/* Tell rb how many elements (or in this case bytes) have been
* written to #buffer. This will also wake any block reader thread */
rb_write_commit(rb, nread);
}
close(cfd);
/* call rb_stop(), so parser thread can exit from blocked rb_read(). */
rb_stop(rb);
/* wait for thread to finish it's job before we destroy rb object */
pthread_join(parser_t, NULL);
/* now we are sure no thread is locked in any rb_* functions, so we can
* safely destroy the object */
rb_destroy(rb);
return 0;
}
|
