-
Notifications
You must be signed in to change notification settings - Fork 21
/
Copy pathvqec_dp_input_shim_api.c
2483 lines (2263 loc) · 90 KB
/
vqec_dp_input_shim_api.c
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
/*
* Copyright (c) 2008-2010 by Cisco Systems, Inc.
* All rights reserved.
*
* The VQE-C input shim receives packets on a STB/host platform and forwards
* them to the core VQE-C data plane module. Since different STB/host
* platforms may have different interfaces with the network, it is expected
* that this file may require customization (e.g. in the context of
* integrating VQE-C into a linux kernel) on a given platform.
*
* The VQE-C input shim receives designated packet streams from the network
* and passes them via APIs to the core VQE-C dataplane APIs. Packets are
* passed from the "output stream" objects on which they are received to
* input stream objects using API function pointers supplied to the
* input shim during stream creation and initialization.
*
* File organization:
* 1. types and #defines
* 2. filter entry functions
* 3. output stream functions
* 4. input shim APIs
*/
#include "vqec_dp_input_shim_private.h"
#include "id_manager.h"
#include "vqec_dp_common.h"
#include "vqec_dp_utils.h"
#include "vqec_dp_debug_utils.h"
#include <utils/zone_mgr.h>
/* included for unit testing purposes */
#ifdef _VQEC_DP_UTEST
#include "test_dp_interposers.h"
#endif
/*
* The macro below can be defined to print out additional debugging info
* during testing/debugging. Since it is not suitable for use in production
* environments (output is only meaningful to a developer and may be on a
* per-packet basis), support for it is not compiled in by default.
*
* Any debugs which are needed in a production environment should use
* VQEC_DP_DEBUG(VQEC_DP_DEBUG_INPUTSHIM, ...) instead.
*/
#if 0
#define VQE_PRINT printk
#define VQEC_DP_INPUT_SHIM_DEBUG(arg...) VQE_PRINT(arg);
#else
#define VQEC_DP_INPUT_SHIM_DEBUG(arg...)
#endif
/*
* Temporary storage used for building debug strings
*/
#define DEBUG_STR_LEN 200
static char s_debug_str[DEBUG_STR_LEN];
static char s_debug_str2[DEBUG_STR_LEN];
#define DEBUG_STR_LEN_SHORT 20
static char s_debug_str_short[DEBUG_STR_LEN_SHORT];
static char s_debug_str_short2[DEBUG_STR_LEN_SHORT];
/* Global parameters cached for socket creation/manipulation */
static uint32_t vqec_dp_input_shim_max_paksize;
static uint32_t vqec_dp_input_shim_pakpool_size;
/* Global status of the input shim */
vqec_dp_input_shim_status_t
vqec_dp_input_shim_status = { TRUE, 0, 0, 0 };
/* Output Stream ID table */
static id_table_key_t vqec_dp_input_shim_os_id_table_key =
ID_MGR_TABLE_KEY_ILLEGAL;
/*
* In case no packets are available from the packet pool, the input shim
* maintains storage for a single packet of its own. This is used primarily
* for draining the sockets if a pak cannot be allocated.
*/
static vqec_pak_t *s_vqec_dp_input_shim_pak = NULL;
/*
* Allocation zones.
*/
static struct vqe_zone *s_vqec_filter_entry_pool = NULL;
static struct vqe_zone *s_vqec_input_shim_os_pool = NULL;
static struct vqe_zone *s_input_shim_pak_pool = NULL;
static struct vqe_zone *s_input_shim_filter_table_pool = NULL;
static vqec_recv_sock_pool_t *s_vqec_recv_sock_pool = NULL;
/*
* Initialize the filter entry table and number of scheduling classes.
*/
uint32_t vqec_dp_input_shim_num_scheduling_classes = 0;
vqec_dp_scheduling_class_t *vqec_dp_input_shim_filter_table = NULL;
/* creates an empty filter entry */
vqec_filter_entry_t *
vqec_dp_input_shim_filter_entry_create (void)
{
vqec_filter_entry_t *filter_entry;
filter_entry = (vqec_filter_entry_t *)
zone_acquire(s_vqec_filter_entry_pool);
if (!filter_entry) {
return NULL;
}
memset(filter_entry, 0, sizeof(vqec_filter_entry_t));
vqec_dp_input_shim_status.num_filters++;
return (filter_entry);
}
/*
* destroys a filter entry, returns resources it holds
*
* @param[in] filter_entry filter entry to be destroyed
*
* NOTE: the filter entry MUST be removed from any databases prior
* to calling this function; otherwise, any pointers to this
* filter entry will become invalid with its destruction
*/
void
vqec_dp_input_shim_filter_entry_destroy (vqec_filter_entry_t *filter_entry)
{
if (!filter_entry) {
return;
}
if (filter_entry->socket) {
vqec_recv_sock_destroy_in_pool(s_vqec_recv_sock_pool,
filter_entry->socket);
}
if (filter_entry->extra_igmp_socket) {
vqec_recv_sock_destroy_in_pool(s_vqec_recv_sock_pool,
filter_entry->extra_igmp_socket);
}
zone_release(s_vqec_filter_entry_pool, filter_entry);
vqec_dp_input_shim_status.num_filters--;
}
/*
* vqec_dp_input_shim_run_service_filter_entry()
*
* Collects packets which have arrived for a bound output stream, and
* forwards them to any connected input streams.
*
* @param[in] filter_entry Filter whose output stream is to be processed
*/
void
vqec_dp_input_shim_run_service_filter_entry (vqec_filter_entry_t *filter_entry)
{
vqec_dp_input_shim_os_t *os;
vqec_pak_t *pak=NULL;
vqec_pak_t *pak_array[VQEC_DP_STREAM_PUSH_VECTOR_PAKS_MAX];
int32_t read_len;
int32_t num_paks_in_array, i;
int32_t num_bytes_in_array;
boolean potentially_more_pkts = TRUE;
boolean static_pkt_in_use = FALSE;
VQEC_DP_ASSERT_FATAL(filter_entry, "inputshim");
os = filter_entry->os;
VQEC_DP_INPUT_SHIM_DEBUG(
"processing filter entry for OS ID '0x%08x'...", os->os_id);
/*
* Avoid processing this filter if its packets will not be
* received by an input stream.
*/
if (!os || !os->is_ops ||
((os->is_capa & VQEC_DP_STREAM_CAPA_PUSH_VECTORED) &&
!os->is_ops->receive_vec) ||
((os->is_capa & VQEC_DP_STREAM_CAPA_PUSH) &&
!os->is_ops->receive)) {
return;
}
/*
* Overall strategy is:
* 1. Read up to a full vector of packets from the socket into an array
* 2. Forward the packet(s) from the array to the Input Stream
* 3. Repeat until the socket runs dry.
*/
do {
/* Initialize array as empty */
num_paks_in_array = 0;
num_bytes_in_array = 0;
/* Read packets from the socket to the array */
while (num_paks_in_array < VQEC_DP_STREAM_PUSH_VECTOR_PAKS_MAX) {
/* alloc without a particle */
pak = vqec_pak_alloc_no_particle();
if (!pak) {
/*
* Out of packet buffers at the moment. Use a static buffer
* to read from the socket, and then discard the packet.
*
* The "socket draining" behavior is primarily targeted to
* help recovery from the scenario where VQE-C packet buffers
* are not available for an extended period of time, in which
* case VQE-C socket buffers would fill up considerably
* (if not drained). By draining the sockets, the likelihood
* of later pushing stale data into the decoder is minimized.
*/
pak = s_vqec_dp_input_shim_pak;
static_pkt_in_use = TRUE;
}
/*
* Set the pak buffer pointer. This is only necessary in userspace,
* as in kernel space, this pointer will be replaced by a pointer
* to the skb's buffer. However, in userspace, this pointer needs
* to be set so vqec_recv_sock_read_pak() knows where to write the
* data that is received from the socket.
*/
pak->buff = (char *)(pak + 1);
read_len = vqec_recv_sock_read_pak(filter_entry->socket,
pak);
if (read_len < 1) {
/*
* The socket has been fully drained. So let's stop trying to
* read packets, and go pass along any we may have collected.
*/
potentially_more_pkts = FALSE;
if (!static_pkt_in_use) {
vqec_pak_free(pak);
}
break;
}
if (static_pkt_in_use) {
/*
* Free any kernel memory associated with the static packet,
* but skip other processing of it. Instead, just go try
* again to allocate a new packet and read from the socket.
*/
vqec_pak_free_skb(pak);
static_pkt_in_use = FALSE;
os->stats.drops++;
/* This is also an overrun in the TR-135 sense */
vqec_dp_input_shim_status.tr135_overruns++;
continue;
}
/* Add the packet to the array. */
pak_array[num_paks_in_array] = pak;
/* Keep track of what's in the array */
num_paks_in_array++;
num_bytes_in_array += vqec_pak_get_content_len(pak);
}
VQEC_DP_INPUT_SHIM_DEBUG(
"collected %u paks\n", num_paks_in_array);
/* Forward the held packets to the Input Stream */
if (num_paks_in_array) {
if (os->is_capa & VQEC_DP_STREAM_CAPA_PUSH_VECTORED) {
VQEC_DP_INPUT_SHIM_DEBUG(" dumping via vector...\n");
if (os->is_ops->receive_vec(os->is_id,
(vqec_pak_t **)pak_array,
num_paks_in_array) !=
VQEC_DP_STREAM_ERR_OK) {
vqec_dp_input_shim_status.num_pkt_errors +=
num_paks_in_array;
}
for (i=0; i<num_paks_in_array; i++) {
vqec_pak_free(pak_array[i]);
}
} else {
VQEC_DP_INPUT_SHIM_DEBUG(" dumping via single...\n");
for (i=0; i<num_paks_in_array; i++) {
if (os->is_ops->receive(os->is_id, pak_array[i]) !=
VQEC_DP_STREAM_ERR_OK) {
vqec_dp_input_shim_status.num_pkt_errors++;
}
vqec_pak_free(pak_array[i]);
}
}
if (VQEC_DP_GET_DEBUG_FLAG(VQEC_DP_DEBUG_COLLECT_STATS)) {
os->stats.packets += num_paks_in_array;
os->stats.bytes += num_bytes_in_array;
}
}
} while (potentially_more_pkts);
}
/*
* vqec_dp_input_shim_run_service()
*
* Service the input shim's output streams.
*
* This API is intended to be invoked periodically, with an "elapsed_time"
* parameter indicating the amount of time (in milliseconds) that has elapsed
* since the previous invocation. All scheduling classes whose intervals
* have terminated within the elapsed time period will be processed (any
* packets that have been buffered for an OS assigned to the scheduling class
* will be forwared to its connected IS).
*
* If invoking this API for the first time, the value passed for the
* "elapsed_time" parameter is not significant--all scheduling classes will
* be processed.
*
* E.g. Assume two scheduling classes:
* class A: interval of 20ms
* class B: interval of 40ms
* These classes will be serviced at the instances shown below,
* if given the sequence of calls to this API:
* absolute time call serviced classes
* ------------- ---------------- ----------------
* t run_service(x): class A, B
* t + 20 run_service(20): class A
* t + 40 run_service(20): class A, B
* t + 60 run_service(20): class A
* t + 70 run_service(10):
* t + 85 run_service(15): class A, B
* t + 100 run_service(15);
* t + 105 run_service(5); class A
* t + 160 run_service(55); class A, B
*
* @param[in] elapsed_time Amount of time which has elapsed since the previous
* call to this function following input shim startup
* (if applicable).
*/
void
vqec_dp_input_shim_run_service (uint16_t elapsed_time)
{
vqec_filter_entry_t *filter_entry;
uint32_t i;
for (i=0; i<vqec_dp_input_shim_num_scheduling_classes; i++) {
if (vqec_dp_input_shim_filter_table[i].remaining) {
/*
* Only deduct the elapsed time if this is not the first service
* call. If it is the first service call, the class will be
* serviced regardless of the elapsed time parameter.
*/
vqec_dp_input_shim_filter_table[i].remaining -= elapsed_time;
}
if (vqec_dp_input_shim_filter_table[i].remaining <= 0) {
VQEC_DP_INPUT_SHIM_DEBUG("\nProcessing filter class %u...\n", i);
VQE_LIST_FOREACH(filter_entry,
&vqec_dp_input_shim_filter_table[i].filters,
list_obj) {
vqec_dp_input_shim_run_service_filter_entry(filter_entry);
}
vqec_dp_input_shim_filter_table[i].remaining =
vqec_dp_input_shim_filter_table[i].interval;
}
}
}
/*
* vqec_dp_input_shim_os_id_to_os()
*
* @param[in] os_id OS ID value
* @return pointer to corresponding os object, or
* NULL if no associated os object
*/
vqec_dp_input_shim_os_t *
vqec_dp_input_shim_os_id_to_os (const vqec_dp_osid_t os_id)
{
uint ret_code;
return (id_to_ptr(os_id, &ret_code, vqec_dp_input_shim_os_id_table_key));
}
/**
* vqec_dp_input_shim_filter_fd()
*
* Return the socket file descriptor of the filter
* for the specified inputshim output stream.
*
* @param[in] os_id OS ID value.
* @return int file descriptor
* -1 if not found.
*/
int vqec_dp_input_shim_filter_fd(vqec_dp_osid_t os_id)
{
vqec_dp_input_shim_os_t *p_os;
p_os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!p_os) {
return -1;
}
if (!p_os->filter_entry ||
!p_os->filter_entry->socket) {
return -1;
}
return (p_os->filter_entry->socket->fd);
}
/*
* vqec_dp_input_shim_os_capa()
*
* Get the capability flags for the output stream. This identifies the
* ways in which the output stream may pass packets to an input stream.
*
* @param[in] os A valid output stream handle.
* @param[out] int32_t Stream's capability flags.
*/
int32_t
vqec_dp_input_shim_os_capa (vqec_dp_osid_t os_id)
{
vqec_dp_input_shim_os_t *os;
os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!os) {
return VQEC_DP_STREAM_CAPA_NULL;
}
return (os->capa);
}
/*
* vqec_dp_input_shim_os_accept_connect()
*
* Connects an output stream to the given input stream. This implies
* caching the input stream operations interface, and delivering datagrams
* to the input stream through that interface upon their arrival.
*
* An OS may only be connected to one IS. If an IS is already connected
* to the given OS, this function will fail. The "disconnect" API may be
* used to disconnect an IS from an OS.
*
* In addition to connecting an OS to an IS, it is necessary to "bind" the OS
* to a filter (which defines the traffic stream of this OS) before traffic
* may be passed from OS -> IS.
*
* Parameters:
* @param[in] os_id A valid handle indicating the output stream which is
* to be connected to an IS.
* @param[in] is_id A valid handle identifying the input stream to be
* connected to the OS.
* @param[in] ops The operations interface of the input stream.
* The caller MUST ensure this memory is not freed or
* overwritten for the lifespan of this output stream,
* as its address will be cached internally.
* This operations interface includes the "receive" API,
* which is used to deliver packets to the connected IS.
* @param[in] encap The expected encapsulation type for traffic passed from
* the OS to IS. The value may represent UDP or RTP;
* and is validated against the encapsulation type that
* was defined for the OS upon its creation. If the
* type encapsulation values do not match, the
* VQEC_DP_STREAM_ERR_ENCAPSMISMATCH error code is
* returned.
* @param[in] req_capa Specifies how the OS will deliver its stream to IS.
*/
vqec_dp_stream_err_t
vqec_dp_input_shim_os_accept_connect (vqec_dp_osid_t os_id,
vqec_dp_isid_t is_id,
const vqec_dp_isops_t *ops,
vqec_dp_encap_type_t encap,
int32_t req_capa)
{
vqec_dp_stream_err_t status = VQEC_DP_STREAM_ERR_OK;
vqec_dp_input_shim_os_t *os;
/* Verify the input shim is operational */
if (vqec_dp_input_shim_status.is_shutdown) {
status = VQEC_DP_STREAM_ERR_SERVICESHUT;
goto done;
}
/* Validate the parameters */
os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!os || !ops || (req_capa > VQEC_DP_STREAM_CAPA_MAX) ||
((req_capa & VQEC_DP_STREAM_CAPA_PUSH_VECTORED) &&
!ops->receive_vec) ||
((req_capa & VQEC_DP_STREAM_CAPA_PUSH) && !ops->receive) ||
((req_capa & VQEC_DP_STREAM_CAPA_PUSH_POLL) &&
!(ops->receive || ops->receive_vec))) {
status = VQEC_DP_STREAM_ERR_INVALIDARGS;
goto done;
}
if (encap != os->encaps) {
status = VQEC_DP_STREAM_ERR_ENCAPSMISMATCH;
goto done;
}
if (req_capa & ~os->capa) {
status = VQEC_DP_STREAM_ERR_NACKCAPA;
goto done;
}
if (os->is_id != VQEC_DP_INVALID_ISID) {
status = VQEC_DP_STREAM_ERR_OSALREADYCONNECTED;
goto done;
}
/* Connection can succeed, store associated IS information */
os->is_id = is_id;
os->is_ops = ops;
os->is_capa = req_capa;
done:
VQEC_DP_DEBUG(VQEC_DP_DEBUG_INPUTSHIM, "%s(os_id=0x%08x, is_id=%u, ops=%p,"
" encap=%s, req_capa=%s)%s\n",
__FUNCTION__, os_id, is_id, ops,
vqec_dp_stream_encap2str(encap, s_debug_str, DEBUG_STR_LEN),
vqec_dp_stream_capa2str(req_capa, s_debug_str2,
DEBUG_STR_LEN),
vqec_dp_stream_err2str_complain_only(status));
return (status);
}
/*
* vqec_dp_input_shim_os_disconnect()
*
* Disconnect the output stream from any input stream that
* it may have been connected to via an earlier connect call.
*
* @param[in] os A valid output stream handle.
*/
void
vqec_dp_input_shim_os_disconnect (vqec_dp_osid_t os_id)
{
vqec_dp_input_shim_os_t *os;
/* Verify the input shim is operational */
if (vqec_dp_input_shim_status.is_shutdown) {
goto done;
}
/* Validate the parameter */
os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!os) {
goto done;
}
/* Disconnect the OS from its IS, if any */
os->is_id = VQEC_DP_INVALID_ISID;
os->is_ops = NULL;
os->is_capa = VQEC_DP_STREAM_CAPA_NULL;
done:
VQEC_DP_DEBUG(VQEC_DP_DEBUG_INPUTSHIM, "%s(os_id=0x%08x)\n",
__FUNCTION__, os_id);
return;
}
/*
* vqec_dp_input_shim_os_bind()
*
* Binds an input filter to an output stream. The input filter identifies
* the set of arriving traffic which constitutes the output stream.
*
* The passed filter must NOT match traffic which also matches any existing
* filter. In other words, the set of traffic streams matched by any two
* filters must be disjoint. If this assumption is not met, the behavior
* of overlapping filters is unspecified.
*
* Source information (ip address and/or port) may be used in filtering
* if requested via the filter's "src_ip_filter" and "src_port_filter"
* fields, with the following caveats for multicast streams:
* 1. requests to filter based on source IP address are supported only
* if the kernel version used supports this capability (true with
* recent kernel versions), and
* 2. requests to filter based on source port are not supported
* and the following caveat for unicast streams:
* 1. if a source port filter is supplied then a source address
* filter must also be supplied
*
* Callers may specify a 0 for the destination port when using a unicast
* destination address. In this case, an unused destination port will
* be allocated and kept open as long as the output stream remains bound.
* The allocated port must be returned back to the caller in *port.
*
* Callers may supply the value INADDR_ANY for the destination IP address.
* The meaning of INADDR_ANY as a destination IP address differs as follows:
*
* If a source address filter and source port filter are NOT supplied,
* then only packets arriving that have filter's destination port and a
* destination IP address that matches the address configured on *any*
* interface on the box will be considered a match.
*
* If a source address filter and source port filter ARE supplied,
* then ONLY packets arriving that have the filter's destination port and
* the "designated" interface's IP address will be considered a match.
* Here, the "designated" interface is chosen via a local route lookup
* (done at the time of filter binding) to see which interface would be
* used if sending to the IP address supplied as the filter's source
* IP address. If the local routing decision were to change such that
* a different interface would be used when sending a packet to the
* filter's source IP address, then the behavior of the filter is
* unspecified.
*
* The scheduling class of a filter is used in conjunction with the
* vqec_dp_input_shim_startup() and vqec_dp_input_shim_run_service() APIs.
* When the vqec_dp_input_shim_run_service() API is called with an
* "elapsed_time" parameter, the set of scheduling classes that need to be
* serviced is determined. If the supplied filter has been assigned to
* such a scheduling class, it will be serviced in that context.
*
* @param[in] os_id A valid output stream handle.
* @param[in] fil Pointer to the input filter (with IP
* addresses and ports in network byte order)
* @param[out] *port If the filter destination port is 0,
* an unused port will be allocated and
* returned to the user in this parameter.
* (in network byte order)
* @param[out] *sock_fd If the socket is used for the filter,
* return the fd of that socket for use
* in NAT injection.
* @param[in] so_rcvbuf Maximum number of arriving bytes that
* may be buffered by the OS, prior to it
* being serviced. Arriving bytes in excess
* of this limit will be dropped.
* @param[in] scheduling_class Associates the filter with the given
* scheduling_class (e.g. 0,1,2,3...)
* See the vqec_dp_input_shim_run_service()
* function header for details on how this
* is used.
* @param[in] xmit_dscp_value DSCP value to be used for all pkts
* transmitted to this stream's source,
* if any. Packets can be transmitted
* to the stream's source via the
* returned "sock_fd" value. The API
* enforces the requested dscp_value by
* assigning the DSCP as a socket option.
* If an elevated DSCP cannot be assigned,
* an error is logged, and the
* default/best-effort DSCP is used.
* @param[out] vqec_dp_stream_err_t Returns STREAM_ERR_OK on success.
*/
vqec_dp_stream_err_t
vqec_dp_input_shim_os_bind (vqec_dp_osid_t os_id,
vqec_dp_input_filter_t *fil,
uint16_t *port,
int *sock_fd,
uint32_t so_rcvbuf,
uint32_t scheduling_class,
uint8_t xmit_dscp_value)
{
in_addr_t rcv_if_address;
in_addr_t mcast_group;
struct in_addr saddr;
vqec_dp_stream_err_t status = VQEC_DP_STREAM_ERR_OK;
vqec_dp_input_shim_os_t *os;
vqec_filter_entry_t *filter_entry = NULL;
vqec_recv_sock_t *vqec_recv_sock = NULL;
vqec_recv_sock_t *extra_igmp_sock = NULL;
int modified_so_rcvbuf;
if (vqec_dp_input_shim_status.is_shutdown) {
status = VQEC_DP_STREAM_ERR_SERVICESHUT;
goto done;
}
/*
* Validate parameters
*
* Checks for any of the following invalid situations:
* 1. Supplied OS ID doesn't map to a valid OS
* 2. Filter is not supplied
* 3. Filter destination port is zero and port parameter (used to
* return an internally allocated port) is NULL
* 4. Scheduling class exceeds the limit configured at startup
*/
os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!os ||
!fil ||
(!port && !ntohs(fil->u.ipv4.dst_port)) ||
(scheduling_class >= vqec_dp_input_shim_num_scheduling_classes)) {
status = VQEC_DP_STREAM_ERR_INVALIDARGS;
goto done;
}
/*
* Check for any of the following unsupported filter situations:
* 1. Filter protocol is not UDP
* 2. Filter destination addr is multicast and destination port is zero
* 3. Filter destination addr is multicast and a src port filter is
* requested
* 4. Filter destination is unicast, a src port filter is requested,
* but a src addr filter is not
*/
if (fil->proto != INPUT_FILTER_PROTO_UDP ||
(IN_MULTICAST(ntohl(fil->u.ipv4.dst_ip)) &&
!ntohs(fil->u.ipv4.dst_port)) ||
(IN_MULTICAST(ntohl(fil->u.ipv4.dst_ip)) &&
fil->u.ipv4.src_port_filter) ||
(!IN_MULTICAST(ntohl(fil->u.ipv4.dst_ip)) &&
(fil->u.ipv4.src_port_filter && !fil->u.ipv4.src_ip_filter))) {
status = VQEC_DP_STREAM_ERR_FILTERUNSUPPORTED;
goto done;
}
/* Verify filter is not already bound (caller must unbind first if so) */
if (os->filter_entry) {
status = VQEC_DP_STREAM_ERR_OSALREADYBOUND;
goto done;
}
/* Create a filter entry to store the binding */
filter_entry = vqec_dp_input_shim_filter_entry_create();
if (!filter_entry) {
status = VQEC_DP_STREAM_ERR_NOMEMORY;
goto done;
}
/* Validation succeeded: create and bind the socket */
if (IN_MULTICAST(ntohl(fil->u.ipv4.dst_ip))) {
rcv_if_address = fil->u.ipv4.dst_ifc_ip;
mcast_group = fil->u.ipv4.dst_ip;
} else {
rcv_if_address = fil->u.ipv4.dst_ip;
mcast_group = 0;
}
vqec_recv_sock = vqec_recv_sock_create_in_pool(
s_vqec_recv_sock_pool,
"",
rcv_if_address,
fil->u.ipv4.dst_port,
mcast_group,
FALSE,
so_rcvbuf,
xmit_dscp_value);
if (!vqec_recv_sock) {
status = VQEC_DP_STREAM_ERR_INTERNAL;
goto done;
}
filter_entry->socket = vqec_recv_sock;
/*
* The reference implementation of the input and output shim share the
* memory allocated for the vqec_recv_socket. The vqec_recv_socket is used
* to filter packets (from above) using the IP stack at the socket layer
* within the input shim and then receive them into this socket memory.
* The same packet data memory is used throughout the rest of VQE-C,
* including the output shim.
*
* Socket accounting normally accounts for buffering via the setsockopt()
* function call. However, this call only permits the socket buffer to be
* resized to a maximum of 1 Mbyte, which is not enough to encompass the
* desired VQE-C packet cache. Rather than requiring the modification of
* kernel source to expand the amount of memory that a socket can use, the
* code below directly modifies the socket parameter that is normally
* modified by the setsockopt() processing.
*/
/* adjust SO_RCVBUF without restrictions (works in kernel-space only) */
modified_so_rcvbuf = so_rcvbuf +
(vqec_dp_input_shim_pakpool_size *
(vqec_dp_input_shim_max_paksize + vqec_recv_sock_get_pak_overhead()));
(void)vqec_recv_sock_set_so_rcvbuf(vqec_recv_sock,
(so_rcvbuf ? TRUE : FALSE),
modified_so_rcvbuf);
if (IN_MULTICAST(ntohl(fil->u.ipv4.dst_ip))) {
/*
* multicast destination:
* 1. if requested, apply source filter of (addr)
* 2. perform multicast join to receive traffic
*/
saddr.s_addr = (fil->u.ipv4.src_ip_filter ?
fil->u.ipv4.src_ip : INADDR_ANY);
if (!vqec_recv_sock_mcast_join(vqec_recv_sock, saddr)) {
status = VQEC_DP_STREAM_ERR_INTERNAL;
goto done;
}
} else {
/* unicast destination:
* 1. if requested, apply source filter using (addr, port)
* 2. if requested, join extra igmp ip to stop previous mcast
*/
if (fil->u.ipv4.src_ip_filter || fil->u.ipv4.src_port_filter) {
if (!vqec_recv_sock_connect(vqec_recv_sock,
fil->u.ipv4.src_ip_filter,
fil->u.ipv4.src_ip,
fil->u.ipv4.src_port_filter,
fil->u.ipv4.src_port)) {
status = VQEC_DP_STREAM_ERR_INTERNAL;
goto done;
}
}
/* If using an extra igmp multicast ip, bind to it here */
if (fil->u.ipv4.rcc_extra_igmp_ip) {
extra_igmp_sock = vqec_recv_sock_create_in_pool(
s_vqec_recv_sock_pool,
"",
fil->u.ipv4.dst_ifc_ip,
fil->u.ipv4.dst_port,
fil->u.ipv4.rcc_extra_igmp_ip,
FALSE,
0 /* so_rcvbuf */,
0 /* xmit_dscp_value */);
if (!extra_igmp_sock) {
/* Allocation failure */
status = VQEC_DP_STREAM_ERR_INTERNAL;
goto done;
}
if (!vqec_recv_sock_mcast_join(extra_igmp_sock, (struct in_addr){0})) {
vqec_recv_sock_destroy_in_pool(s_vqec_recv_sock_pool,
extra_igmp_sock);
status = VQEC_DP_STREAM_ERR_INTERNAL;
goto done;
}
filter_entry->extra_igmp_socket = extra_igmp_sock;
}
}
/* Success: assign the remaining parameters to the filter entry */
filter_entry->scheduling_class = scheduling_class;
filter_entry->xmit_dscp_value = xmit_dscp_value;
memcpy(&filter_entry->filter, fil, sizeof(vqec_dp_input_filter_t));
if (!ntohs(filter_entry->filter.u.ipv4.dst_port)) {
filter_entry->filter.u.ipv4.dst_port =
vqec_recv_sock_get_port(vqec_recv_sock);
}
filter_entry->os = os;
/* Insert the filter entry into the filter table */
VQE_LIST_INSERT_HEAD(
&vqec_dp_input_shim_filter_table[scheduling_class].filters,
filter_entry, list_obj);
/*
* Filter entry is now in bind-committed state, i.e., it is on the
* scheduler list, and must be removed off that list at unbind.
*/
filter_entry->committed = TRUE;
/* Link to the filter from the OS */
os->filter_entry = filter_entry;
done:
if (status != VQEC_DP_STREAM_ERR_OK) {
vqec_dp_input_shim_filter_entry_destroy(filter_entry);
if (port) {
*port = htons(0);
}
} else {
if (!ntohs(fil->u.ipv4.dst_port)) {
*port = vqec_recv_sock_get_port(vqec_recv_sock);
}
}
if (port) {
snprintf(s_debug_str_short, DEBUG_STR_LEN_SHORT, "%u", *port);
} else {
snprintf(s_debug_str_short, DEBUG_STR_LEN_SHORT, "NULL");
}
VQEC_DP_DEBUG(VQEC_DP_DEBUG_INPUTSHIM, "%s(os_id=0x%08x, filter=%s, "
"port=%s, so_rcvbuf=%u, "
"scheduling_class=%u, xmit_dscp_value=%u)%s\n",
__FUNCTION__, os_id,
vqec_dp_input_filter_to_str(fil, TRUE, TRUE, TRUE),
s_debug_str_short, so_rcvbuf,
scheduling_class, xmit_dscp_value,
vqec_dp_stream_err2str_complain_only(status));
return (status);
}
/*
* vqec_dp_input_shim_os_unbind()
*
* Removes any previous binding for the given output stream.
*
* @param[in] os_id A valid output stream handle.
*/
void
vqec_dp_input_shim_os_unbind (vqec_dp_osid_t os_id)
{
vqec_dp_input_shim_os_t *os;
if (vqec_dp_input_shim_status.is_shutdown) {
goto done;
}
os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!os || !os->filter_entry) {
goto done;
}
if (os->filter_entry->committed) {
/* Entry is on the scheduler list iff the entry was committed. */
VQE_LIST_REMOVE(os->filter_entry, list_obj);
}
vqec_dp_input_shim_filter_entry_destroy(os->filter_entry);
os->filter_entry = NULL;
done:
VQEC_DP_DEBUG(VQEC_DP_DEBUG_INPUTSHIM, "%s(os_id=0x%08x)\n",
__FUNCTION__, os_id);
return;
}
/*
* vqec_dp_input_shim_os_get_status()
*
* Get stream status for an output stream. This includes the number of
* connected input streams, the filter binding (if any), and
* statistics. An implementation may choose not to keep per output
* statistics. The statistics should be reset to 0 for that case.
*
* @param[in] os A valid output stream handle.
* @param[out] status Status structure for the result.
* @param[out] vqec_dp_stream_err_t Returns STREAM_ERR_OK on success.
*/
vqec_dp_stream_err_t
vqec_dp_input_shim_os_get_status (vqec_dp_osid_t os_id,
vqec_dp_os_status_t *stat)
{
vqec_dp_input_shim_os_t *os;
vqec_dp_stream_err_t status = VQEC_DP_STREAM_ERR_OK;
/* Validate parameters */
os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!os || !stat) {
status = VQEC_DP_STREAM_ERR_INVALIDARGS;
goto done;
}
memset(stat, 0, sizeof(vqec_dp_os_status_t));
stat->is_cnt = 1;
if (os->filter_entry) {
memcpy(&stat->filter,
&os->filter_entry->filter, sizeof(vqec_dp_input_filter_t));
}
memcpy(&stat->stats, &os->stats, sizeof(vqec_dp_stream_stats_t));
done:
return status;
}
/*
* vqec_dp_input_shim_os_get_connected_is()
*
* Retrieves the set of input streams associated with the given
* output stream.
*
* Callers supply an isid_array (of size isid_array_len) into which
* IS IDs for OS are copied. The number of IS IDs copied into the
* array is bound by the array size (specified by isid_array_len),
* and returned via the num_isids parameter.
*
* If there are more IS IDs for the OS that were not copied into
* the isid_array (due to space limitations), then the more parameter
* will be set to TRUE upon return of the function. The caller may
* then call the function again, with is_id_last assigned the ID value
* of the last ID returned in the previous call. (If this is the first
* call, is_id_last should be VQEC_DP_INVALID_ISID to start from the
* beginning of the list.) Should the OS to IS stream mapping change
* between individual calls to this function, callers may not retrieve
* an atomic snapshot of the OS to IS mapping.
*
* @param[in] os_id The OS whose ISes are to be retrieved
* @param[in] is_id_last The last IS retrieved from the previous call
* of a sequence (which returned more==TRUE), or
* VQEC_DP_INVALID_ISID for the initial call
* (to retrieve IS IDs starting at the beginning)
* @param[out] isid_array Array into which the ISes of os_id are copied
* @param[in] isid_array_len Size of isid_array (in elements)
* @param[out] num_isids Number of IS IDs copied into isid_array
* @param[out] more TRUE if there are more IS IDs to be retrieved
* for OS, or FALSE otherwise.
* @param[out] vqec_dp_stream_err_t Success/failure status of request
*/
vqec_dp_stream_err_t
vqec_dp_input_shim_os_get_connected_is (vqec_dp_osid_t os_id,
vqec_dp_isid_t is_id_last,
vqec_dp_isid_t *isid_array,
uint32_t isid_array_len,
uint32_t *num_isids,
boolean *more)
{
vqec_dp_input_shim_os_t *os;
vqec_dp_stream_err_t status;
/* Validate parameters */
os = vqec_dp_input_shim_os_id_to_os(os_id);
if (!os || !isid_array || !isid_array_len || !num_isids || !more) {
status = VQEC_DP_STREAM_ERR_INVALIDARGS;
goto done;
}
/*
* Output Streams can only have 0 or 1 connected input stream right now.
* Copy its IS ID into the first position in the array.
*/
if ((is_id_last == VQEC_DP_INVALID_ISID) &&
(os->is_id != VQEC_DP_INVALID_ISID)) {
isid_array[0] = os->is_id;
*num_isids = 1;
} else {
*num_isids = 0;
}
*more = FALSE;
status = VQEC_DP_STREAM_ERR_OK;
done:
return (status);
}