Line data Source code
1 : /*
2 : * Copyright (c) 2015 Semihalf. All rights reserved.
3 : */
4 :
5 : #include <vr_dpdk.h>
6 : #include <vr_fragment.h>
7 : #include <vr_os.h>
8 : #include <vr_packet.h>
9 :
10 : /**
11 : * @name Private variables
12 : * @{
13 : */
14 :
15 : /* Hash table buckets used by the assembler */
16 : struct fragment_bucket {
17 : struct vr_fragment *frag_list;
18 : };
19 : static struct fragment_bucket **assembler_table;
20 :
21 : /* Per CPU queues used to enqueue packets for the assembler */
22 : struct per_cpu_fragment_queue {
23 : struct vr_fragment_queue queue;
24 : };
25 : static struct per_cpu_fragment_queue *per_cpu_queues;
26 :
27 : static int assembler_scan_index;
28 : static int assembler_scan_thresh = 1024;
29 :
30 : /** @} */
31 :
32 : /**
33 : * @name Private functions
34 : * @{
35 : */
36 : static void
37 21 : dpdk_fragment_assemble_queue(void *arg)
38 : {
39 21 : struct per_cpu_fragment_queue *queue = (struct per_cpu_fragment_queue *)arg;
40 21 : struct vr_fragment_queue *fq = &queue->queue;
41 21 : vr_fragment_assemble_queue(fq);
42 21 : }
43 :
44 : static void
45 53 : dpdk_assembler_table_exit(void)
46 : {
47 : int i;
48 :
49 53 : vr_assembler_table_scan_exit();
50 :
51 53 : if (assembler_table) {
52 159 : for (i = 0; i < vr_dpdk.nb_fwd_lcores; ++i) {
53 106 : if (assembler_table[i] != NULL) {
54 106 : vr_free(assembler_table[i], VR_ASSEMBLER_TABLE_OBJECT);
55 106 : assembler_table[i] = NULL;
56 : }
57 : }
58 :
59 53 : vr_free(assembler_table, VR_ASSEMBLER_TABLE_OBJECT);
60 53 : assembler_table = NULL;
61 : }
62 :
63 53 : return;
64 : }
65 :
66 : static int
67 53 : dpdk_assembler_table_init(void)
68 : {
69 : unsigned int size;
70 : int i;
71 :
72 : /* Allocate array of pointers to the assembler tables for each fwd lcore */
73 53 : size = sizeof(struct fragment_bucket *) * vr_dpdk.nb_fwd_lcores;
74 53 : assembler_table = vr_zalloc(size, VR_ASSEMBLER_TABLE_OBJECT);
75 53 : if (!assembler_table) {
76 0 : RTE_LOG(ERR, VROUTER, "%s:%d Allocation for %u failed\n",
77 : __FUNCTION__, __LINE__, size);
78 0 : return -ENOMEM;
79 : }
80 :
81 : /* Now allocate the assembler tables for each fwd lcore */
82 53 : size = sizeof(struct fragment_bucket) * VR_ASSEMBLER_BUCKET_COUNT;
83 159 : for (i = 0; i < vr_dpdk.nb_fwd_lcores; ++i) {
84 106 : assembler_table[i] = vr_zalloc(size, VR_ASSEMBLER_TABLE_OBJECT);
85 106 : if (!assembler_table[i]) {
86 0 : RTE_LOG(ERR, VROUTER, "%s:%d Allocation for %u failed\n",
87 : __FUNCTION__, __LINE__, size);
88 0 : return -ENOMEM;
89 : }
90 : }
91 :
92 : /* Intentionally the vr_assembler_table_scan_init() is not called here as
93 : * it would set up timers on the timer lcore. For the timers the forwarding
94 : * lcores are used, therefore allowing for complete lock elimination. */
95 :
96 53 : return 0;
97 : }
98 :
99 : static int
100 53 : dpdk_fragment_queue_init(void)
101 : {
102 : unsigned int size;
103 :
104 53 : size = sizeof(struct per_cpu_fragment_queue) * vr_dpdk.nb_fwd_lcores;
105 53 : per_cpu_queues = vr_zalloc(size, VR_FRAGMENT_QUEUE_OBJECT);
106 :
107 53 : if (!per_cpu_queues) {
108 0 : RTE_LOG(ERR, VROUTER, "%s: Error allocating fragmentation queues\n",
109 : __func__);
110 0 : return -ENOMEM;
111 : }
112 :
113 53 : return 0;
114 : }
115 :
116 : static void
117 53 : dpdk_fragment_queue_exit(void)
118 : {
119 : int i;
120 :
121 53 : if (!per_cpu_queues) {
122 0 : return;
123 : }
124 :
125 159 : for (i = 0; i < vr_dpdk.nb_fwd_lcores; ++i) {
126 106 : if (per_cpu_queues[i].queue.vfq_tail != NULL)
127 0 : vr_fragment_queue_free(&per_cpu_queues[i].queue);
128 : }
129 :
130 53 : vr_free(per_cpu_queues, VR_FRAGMENT_QUEUE_OBJECT);
131 53 : per_cpu_queues = NULL;
132 : }
133 :
134 : /** @} */
135 :
136 : /**
137 : * @name Public functions
138 : * @{
139 : */
140 :
141 : /**
142 : * Enqueue a packet to per cpu queue and schedule work to
143 : * enqueue/process it to assembler_table[]
144 : *
145 : * Executed only from the forwarding lcores.
146 : */
147 : int
148 21 : dpdk_fragment_assembler_enqueue(struct vrouter *router, struct vr_packet *pkt,
149 : struct vr_forwarding_md *fmd)
150 : {
151 : int ret;
152 : unsigned int cpu;
153 : struct vr_dpdk_lcore *lcore;
154 :
155 21 : cpu = vr_get_cpu();
156 21 : if (cpu >= vr_num_cpus || cpu < VR_DPDK_FWD_LCORE_ID) {
157 0 : RTE_LOG(ERR, VROUTER, "%s:%d Enqueue to the assembler can only be "
158 : "done on forwarding lcores, not on cpu %u\n",
159 : __FUNCTION__, __LINE__, cpu);
160 0 : PKT_LOG(VP_DROP_FRAGMENTS, pkt, 0, VR_DPDK_FRAGMENT_ASSEMBLER_C, __LINE__);
161 0 : vr_pfree(pkt, VP_DROP_FRAGMENTS);
162 0 : return -EINVAL;
163 : }
164 :
165 : /* Enqueue the packet to per CPU queue */
166 21 : ret = vr_fragment_enqueue(router,
167 21 : &per_cpu_queues[cpu - VR_DPDK_FWD_LCORE_ID].queue, pkt, fmd);
168 :
169 21 : if (!ret) {
170 21 : lcore = vr_dpdk.lcores[cpu];
171 : /* Schedule work to enqueue/process by the assembler */
172 21 : vr_dpdk_lcore_schedule_assembler_work(lcore,
173 : dpdk_fragment_assemble_queue,
174 21 : &per_cpu_queues[cpu - VR_DPDK_FWD_LCORE_ID].queue);
175 : }
176 :
177 21 : return 0;
178 : }
179 :
180 : /**
181 : * Assemble packet
182 : */
183 : void
184 21 : dpdk_fragment_sync_assemble(struct vr_fragment_queue_element *vfqe)
185 : {
186 : uint32_t hash, index;
187 : unsigned int cpu;
188 : struct fragment_bucket *bucket;
189 :
190 21 : cpu = vr_get_cpu() - VR_DPDK_FWD_LCORE_ID;
191 21 : assert(cpu >= 0 && cpu < (vr_num_cpus - VR_DPDK_FWD_LCORE_ID));
192 :
193 21 : hash = vr_fragment_get_hash(&vfqe->fqe_pnode);
194 21 : index = (hash % VR_ASSEMBLER_BUCKET_COUNT);
195 21 : bucket = &assembler_table[cpu][index];
196 :
197 21 : vr_fragment_assemble(&bucket->frag_list, vfqe);
198 21 : }
199 :
200 : /**
201 : * A callback for timeouts.
202 : * - Clean stale entries in assembler_table[][]
203 : *
204 : * Called on forwarding lcores only.
205 : */
206 : void
207 1621820 : dpdk_fragment_assembler_table_scan(void *arg)
208 : {
209 1621820 : unsigned int i, j, scanned = 0;
210 : unsigned int cpu;
211 : struct fragment_bucket *vfb;
212 :
213 1621820 : cpu = vr_get_cpu() - VR_DPDK_FWD_LCORE_ID;
214 1621176 : assert(cpu >= 0 && cpu < (vr_num_cpus - VR_DPDK_FWD_LCORE_ID));
215 :
216 1621180 : i = assembler_scan_index;
217 1648487869 : for (j = 0; j < VR_ASSEMBLER_BUCKET_COUNT; j++) {
218 1646866689 : vfb = &assembler_table[cpu][(i + j) % VR_ASSEMBLER_BUCKET_COUNT];
219 :
220 1646866689 : if (vfb->frag_list)
221 2369 : scanned += vr_assembler_table_scan(&vfb->frag_list);
222 :
223 1646866689 : if (scanned > assembler_scan_thresh) {
224 0 : j++;
225 0 : break;
226 : }
227 : }
228 :
229 1621180 : assembler_scan_index = (i + j) % VR_ASSEMBLER_BUCKET_COUNT;
230 1621180 : return;
231 : }
232 :
233 : /**
234 : * Init the fragment assembler.
235 : *
236 : * Called only once during initialization on the master lcore only.
237 : */
238 : int
239 53 : dpdk_fragment_assembler_init(void)
240 : {
241 : int ret;
242 :
243 53 : if ((ret = dpdk_fragment_queue_init()))
244 0 : return ret;
245 :
246 53 : if ((ret = dpdk_assembler_table_init()))
247 0 : return ret;
248 :
249 53 : return 0;
250 : }
251 :
252 : /**
253 : * Exit the fragment assembler and clean up all related data.
254 : *
255 : * Called only once on the master lcore after the forwarding lcores exited.
256 : * Therefore we can securely free everything inside dpdk_fragment_queue_exit()
257 : * and dpdk_assembler_table_exit(). If someone adds call to this in other
258 : * place, she/he has to take care of possible races between the master and
259 : * forwarding lcores as assembler tasks might be using the fragment table or
260 : * per cpu queues at the same time.
261 : */
262 : void
263 53 : dpdk_fragment_assembler_exit(void)
264 : {
265 53 : dpdk_fragment_queue_exit();
266 53 : dpdk_assembler_table_exit();
267 53 : }
268 :
269 : /** @} */
|
