-
Notifications
You must be signed in to change notification settings - Fork 498
/
module.c
14020 lines (12873 loc) · 569 KB
/
module.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) 2016, Salvatore Sanfilippo <antirez at gmail dot com>
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
/* --------------------------------------------------------------------------
* Modules API documentation information
*
* The comments in this file are used to generate the API documentation on the
* website.
*
* Each function starting with VM_ and preceded by a block comment is included
* in the API documentation. To hide a VM_ function, put a blank line between
* the comment and the function definition or put the comment inside the
* function body.
*
* The functions are divided into sections. Each section is preceded by a
* documentation block, which is comment block starting with a markdown level 2
* heading, i.e. a line starting with ##, on the first line of the comment block
* (with the exception of a ----- line which can appear first). Other comment
* blocks, which are not intended for the modules API user, such as this comment
* block, do NOT start with a markdown level 2 heading, so they are included in
* the generated a API documentation.
*
* The documentation comments may contain markdown formatting. Some automatic
* replacements are done, such as the replacement of RM with ValkeyModule in
* function names. For details, see the script src/modules/gendoc.rb.
* -------------------------------------------------------------------------- */
#include "server.h"
#include "cluster.h"
#include "heavyloadlog.h"
#include "rdb.h"
#include "monotonic.h"
#include "script.h"
#include "call_reply.h"
#include "hdr_histogram.h"
#include "crc16_slottable.h"
#include "valkeymodule.h"
#include <dlfcn.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <fcntl.h>
#include <string.h>
/* --------------------------------------------------------------------------
* Private data structures used by the modules system. Those are data
* structures that are never exposed to Modules, if not as void
* pointers that have an API the module can call with them)
* -------------------------------------------------------------------------- */
struct ValkeyModuleInfoCtx {
struct ValkeyModule *module;
dict *requested_sections;
sds info; /* info string we collected so far */
int sections; /* number of sections we collected so far */
int in_section; /* indication if we're in an active section or not */
int in_dict_field; /* indication that we're currently appending to a dict */
};
/* This represents a shared API. Shared APIs will be used to populate
* the server.sharedapi dictionary, mapping names of APIs exported by
* modules for other modules to use, to their structure specifying the
* function pointer that can be called. */
struct ValkeyModuleSharedAPI {
void *func;
ValkeyModule *module;
};
typedef struct ValkeyModuleSharedAPI ValkeyModuleSharedAPI;
dict *modules; /* Hash table of modules. SDS -> ValkeyModule ptr.*/
/* Entries in the context->amqueue array, representing objects to free
* when the callback returns. */
struct AutoMemEntry {
void *ptr;
int type;
};
/* AutoMemEntry type field values. */
#define VALKEYMODULE_AM_KEY 0
#define VALKEYMODULE_AM_STRING 1
#define VALKEYMODULE_AM_REPLY 2
#define VALKEYMODULE_AM_FREED 3 /* Explicitly freed by user already. */
#define VALKEYMODULE_AM_DICT 4
#define VALKEYMODULE_AM_INFO 5
/* The pool allocator block. Modules can allocate memory via this special
* allocator that will automatically release it all once the callback returns.
* This means that it can only be used for ephemeral allocations. However
* there are two advantages for modules to use this API:
*
* 1) The memory is automatically released when the callback returns.
* 2) This allocator is faster for many small allocations since whole blocks
* are allocated, and small pieces returned to the caller just advancing
* the index of the allocation.
*
* Allocations are always rounded to the size of the void pointer in order
* to always return aligned memory chunks. */
#define VALKEYMODULE_POOL_ALLOC_MIN_SIZE (1024*8)
#define VALKEYMODULE_POOL_ALLOC_ALIGN (sizeof(void*))
typedef struct ValkeyModulePoolAllocBlock {
uint32_t size;
uint32_t used;
struct ValkeyModulePoolAllocBlock *next;
char memory[];
} ValkeyModulePoolAllocBlock;
/* This structure represents the context in which modules operate.
* Most APIs module can access, get a pointer to the context, so that the API
* implementation can hold state across calls, or remember what to free after
* the call and so forth.
*
* Note that not all the context structure is always filled with actual values
* but only the fields needed in a given context. */
struct ValkeyModuleBlockedClient;
struct ValkeyModuleUser;
struct ValkeyModuleCtx {
void *getapifuncptr; /* NOTE: Must be the first field. */
struct ValkeyModule *module; /* Module reference. */
client *client; /* Client calling a command. */
struct ValkeyModuleBlockedClient *blocked_client; /* Blocked client for
thread safe context. */
struct AutoMemEntry *amqueue; /* Auto memory queue of objects to free. */
int amqueue_len; /* Number of slots in amqueue. */
int amqueue_used; /* Number of used slots in amqueue. */
int flags; /* VALKEYMODULE_CTX_... flags. */
void **postponed_arrays; /* To set with VM_ReplySetArrayLength(). */
int postponed_arrays_count; /* Number of entries in postponed_arrays. */
void *blocked_privdata; /* Privdata set when unblocking a client. */
ValkeyModuleString *blocked_ready_key; /* Key ready when the reply callback
gets called for clients blocked
on keys. */
/* Used if there is the VALKEYMODULE_CTX_KEYS_POS_REQUEST or
* VALKEYMODULE_CTX_CHANNEL_POS_REQUEST flag set. */
getKeysResult *keys_result;
struct ValkeyModulePoolAllocBlock *pa_head;
long long next_yield_time;
const struct ValkeyModuleUser *user; /* ValkeyModuleUser commands executed via
VM_Call should be executed as, if set */
};
typedef struct ValkeyModuleCtx ValkeyModuleCtx;
#define VALKEYMODULE_CTX_NONE (0)
#define VALKEYMODULE_CTX_AUTO_MEMORY (1<<0)
#define VALKEYMODULE_CTX_KEYS_POS_REQUEST (1<<1)
#define VALKEYMODULE_CTX_BLOCKED_REPLY (1<<2)
#define VALKEYMODULE_CTX_BLOCKED_TIMEOUT (1<<3)
#define VALKEYMODULE_CTX_THREAD_SAFE (1<<4)
#define VALKEYMODULE_CTX_BLOCKED_DISCONNECTED (1<<5)
#define VALKEYMODULE_CTX_TEMP_CLIENT (1<<6) /* Return client object to the pool
when the context is destroyed */
#define VALKEYMODULE_CTX_NEW_CLIENT (1<<7) /* Free client object when the
context is destroyed */
#define VALKEYMODULE_CTX_CHANNELS_POS_REQUEST (1<<8)
#define VALKEYMODULE_CTX_COMMAND (1<<9) /* Context created to serve a command from call() or AOF (which calls cmd->proc directly) */
/* This represents a key opened with VM_OpenKey(). */
struct ValkeyModuleKey {
ValkeyModuleCtx *ctx;
serverDb *db;
robj *key; /* Key name object. */
robj *value; /* Value object, or NULL if the key was not found. */
void *iter; /* Iterator. */
int mode; /* Opening mode. */
union {
struct {
/* List, use only if value->type == OBJ_LIST */
listTypeEntry entry; /* Current entry in iteration. */
long index; /* Current 0-based index in iteration. */
} list;
struct {
/* Zset iterator, use only if value->type == OBJ_ZSET */
uint32_t type; /* VALKEYMODULE_ZSET_RANGE_* */
zrangespec rs; /* Score range. */
zlexrangespec lrs; /* Lex range. */
uint32_t start; /* Start pos for positional ranges. */
uint32_t end; /* End pos for positional ranges. */
void *current; /* Zset iterator current node. */
int er; /* Zset iterator end reached flag
(true if end was reached). */
} zset;
struct {
/* Stream, use only if value->type == OBJ_STREAM */
streamID currentid; /* Current entry while iterating. */
int64_t numfieldsleft; /* Fields left to fetch for current entry. */
int signalready; /* Flag that signalKeyAsReady() is needed. */
} stream;
} u;
};
/* ValkeyModuleKey 'ztype' values. */
#define VALKEYMODULE_ZSET_RANGE_NONE 0 /* This must always be 0. */
#define VALKEYMODULE_ZSET_RANGE_LEX 1
#define VALKEYMODULE_ZSET_RANGE_SCORE 2
#define VALKEYMODULE_ZSET_RANGE_POS 3
/* Function pointer type of a function representing a command inside
* a module. */
struct ValkeyModuleBlockedClient;
typedef int (*ValkeyModuleCmdFunc) (ValkeyModuleCtx *ctx, void **argv, int argc);
typedef int (*ValkeyModuleAuthCallback)(ValkeyModuleCtx *ctx, void *username, void *password, ValkeyModuleString **err);
typedef void (*ValkeyModuleDisconnectFunc) (ValkeyModuleCtx *ctx, struct ValkeyModuleBlockedClient *bc);
/* This struct holds the information about a command registered by a module.*/
struct ValkeyModuleCommand {
struct ValkeyModule *module;
ValkeyModuleCmdFunc func;
struct serverCommand *rediscmd;
};
typedef struct ValkeyModuleCommand ValkeyModuleCommand;
#define VALKEYMODULE_REPLYFLAG_NONE 0
#define VALKEYMODULE_REPLYFLAG_TOPARSE (1<<0) /* Protocol must be parsed. */
#define VALKEYMODULE_REPLYFLAG_NESTED (1<<1) /* Nested reply object. No proto
or struct free. */
/* Reply of VM_Call() function. The function is filled in a lazy
* way depending on the function called on the reply structure. By default
* only the type, proto and protolen are filled. */
typedef struct CallReply ValkeyModuleCallReply;
/* Structure to hold the module auth callback & the Module implementing it. */
typedef struct ValkeyModuleAuthCtx {
struct ValkeyModule *module;
ValkeyModuleAuthCallback auth_cb;
} ValkeyModuleAuthCtx;
/* Structure representing a blocked client. We get a pointer to such
* an object when blocking from modules. */
typedef struct ValkeyModuleBlockedClient {
client *client; /* Pointer to the blocked client. or NULL if the client
was destroyed during the life of this object. */
ValkeyModule *module; /* Module blocking the client. */
ValkeyModuleCmdFunc reply_callback; /* Reply callback on normal completion.*/
ValkeyModuleAuthCallback auth_reply_cb; /* Reply callback on completing blocking
module authentication. */
ValkeyModuleCmdFunc timeout_callback; /* Reply callback on timeout. */
ValkeyModuleDisconnectFunc disconnect_callback; /* Called on disconnection.*/
void (*free_privdata)(ValkeyModuleCtx*,void*);/* privdata cleanup callback.*/
void *privdata; /* Module private data that may be used by the reply
or timeout callback. It is set via the
ValkeyModule_UnblockClient() API. */
client *thread_safe_ctx_client; /* Fake client to be used for thread safe
context so that no lock is required. */
client *reply_client; /* Fake client used to accumulate replies
in thread safe contexts. */
int dbid; /* Database number selected by the original client. */
int blocked_on_keys; /* If blocked via VM_BlockClientOnKeys(). */
int unblocked; /* Already on the moduleUnblocked list. */
monotime background_timer; /* Timer tracking the start of background work */
uint64_t background_duration; /* Current command background time duration.
Used for measuring latency of blocking cmds */
} ValkeyModuleBlockedClient;
/* This is a list of Module Auth Contexts. Each time a Module registers a callback, a new ctx is
* added to this list. Multiple modules can register auth callbacks and the same Module can have
* multiple auth callbacks. */
static list *moduleAuthCallbacks;
static pthread_mutex_t moduleUnblockedClientsMutex = PTHREAD_MUTEX_INITIALIZER;
static list *moduleUnblockedClients;
/* Pool for temporary client objects. Creating and destroying a client object is
* costly. We manage a pool of clients to avoid this cost. Pool expands when
* more clients are needed and shrinks when unused. Please see modulesCron()
* for more details. */
static client **moduleTempClients;
static size_t moduleTempClientCap = 0;
static size_t moduleTempClientCount = 0; /* Client count in pool */
static size_t moduleTempClientMinCount = 0; /* Min client count in pool since
the last cron. */
/* We need a mutex that is unlocked / relocked in beforeSleep() in order to
* allow thread safe contexts to execute commands at a safe moment. */
static pthread_mutex_t moduleGIL = PTHREAD_MUTEX_INITIALIZER;
/* Function pointer type for keyspace event notification subscriptions from modules. */
typedef int (*ValkeyModuleNotificationFunc) (ValkeyModuleCtx *ctx, int type, const char *event, ValkeyModuleString *key);
/* Function pointer type for post jobs */
typedef void (*ValkeyModulePostNotificationJobFunc) (ValkeyModuleCtx *ctx, void *pd);
/* Keyspace notification subscriber information.
* See VM_SubscribeToKeyspaceEvents() for more information. */
typedef struct ValkeyModuleKeyspaceSubscriber {
/* The module subscribed to the event */
ValkeyModule *module;
/* Notification callback in the module*/
ValkeyModuleNotificationFunc notify_callback;
/* A bit mask of the events the module is interested in */
int event_mask;
/* Active flag set on entry, to avoid reentrant subscribers
* calling themselves */
int active;
} ValkeyModuleKeyspaceSubscriber;
typedef struct ValkeyModulePostExecUnitJob {
/* The module subscribed to the event */
ValkeyModule *module;
ValkeyModulePostNotificationJobFunc callback;
void *pd;
void (*free_pd)(void*);
int dbid;
} ValkeyModulePostExecUnitJob;
/* The module keyspace notification subscribers list */
static list *moduleKeyspaceSubscribers;
/* The module post keyspace jobs list */
static list *modulePostExecUnitJobs;
/* Data structures related to the exported dictionary data structure. */
typedef struct ValkeyModuleDict {
rax *rax; /* The radix tree. */
} ValkeyModuleDict;
typedef struct ValkeyModuleDictIter {
ValkeyModuleDict *dict;
raxIterator ri;
} ValkeyModuleDictIter;
typedef struct ValkeyModuleCommandFilterCtx {
ValkeyModuleString **argv;
int argv_len;
int argc;
client *c;
} ValkeyModuleCommandFilterCtx;
typedef void (*ValkeyModuleCommandFilterFunc) (ValkeyModuleCommandFilterCtx *filter);
typedef struct ValkeyModuleCommandFilter {
/* The module that registered the filter */
ValkeyModule *module;
/* Filter callback function */
ValkeyModuleCommandFilterFunc callback;
/* VALKEYMODULE_CMDFILTER_* flags */
int flags;
} ValkeyModuleCommandFilter;
/* Registered filters */
static list *moduleCommandFilters;
typedef void (*ValkeyModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
static struct ValkeyModuleForkInfo {
ValkeyModuleForkDoneHandler done_handler;
void* done_handler_user_data;
} moduleForkInfo = {0};
typedef struct ValkeyModuleServerInfoData {
rax *rax; /* parsed info data. */
} ValkeyModuleServerInfoData;
/* Flags for moduleCreateArgvFromUserFormat(). */
#define VALKEYMODULE_ARGV_REPLICATE (1<<0)
#define VALKEYMODULE_ARGV_NO_AOF (1<<1)
#define VALKEYMODULE_ARGV_NO_REPLICAS (1<<2)
#define VALKEYMODULE_ARGV_RESP_3 (1<<3)
#define VALKEYMODULE_ARGV_RESP_AUTO (1<<4)
#define VALKEYMODULE_ARGV_RUN_AS_USER (1<<5)
#define VALKEYMODULE_ARGV_SCRIPT_MODE (1<<6)
#define VALKEYMODULE_ARGV_NO_WRITES (1<<7)
#define VALKEYMODULE_ARGV_CALL_REPLIES_AS_ERRORS (1<<8)
#define VALKEYMODULE_ARGV_RESPECT_DENY_OOM (1<<9)
#define VALKEYMODULE_ARGV_DRY_RUN (1<<10)
#define VALKEYMODULE_ARGV_ALLOW_BLOCK (1<<11)
/* Determine whether the server should signalModifiedKey implicitly.
* In case 'ctx' has no 'module' member (and therefore no module->options),
* we assume default behavior, that is, the server signals.
* (see VM_GetThreadSafeContext) */
#define SHOULD_SIGNAL_MODIFIED_KEYS(ctx) \
((ctx)->module? !((ctx)->module->options & VALKEYMODULE_OPTION_NO_IMPLICIT_SIGNAL_MODIFIED) : 1)
/* Server events hooks data structures and defines: this modules API
* allow modules to subscribe to certain events in the server, such as
* the start and end of an RDB or AOF save, the change of role in replication,
* and similar other events. */
typedef struct ValkeyModuleEventListener {
ValkeyModule *module;
ValkeyModuleEvent event;
ValkeyModuleEventCallback callback;
} ValkeyModuleEventListener;
list *ValkeyModule_EventListeners; /* Global list of all the active events. */
/* Data structures related to the module users */
/* This is the object returned by VM_CreateModuleUser(). The module API is
* able to create users, set ACLs to such users, and later authenticate
* clients using such newly created users. */
typedef struct ValkeyModuleUser {
user *user; /* Reference to the real user */
int free_user; /* Indicates that user should also be freed when this object is freed */
} ValkeyModuleUser;
/* This is a structure used to export some meta-information such as dbid to the module. */
typedef struct ValkeyModuleKeyOptCtx {
struct serverObject *from_key, *to_key; /* Optional name of key processed, NULL when unknown.
In most cases, only 'from_key' is valid, but in callbacks
such as `copy2`, both 'from_key' and 'to_key' are valid. */
int from_dbid, to_dbid; /* The dbid of the key being processed, -1 when unknown.
In most cases, only 'from_dbid' is valid, but in callbacks such
as `copy2`, 'from_dbid' and 'to_dbid' are both valid. */
} ValkeyModuleKeyOptCtx;
/* Data structures related to module configurations */
/* The function signatures for module config get callbacks. These are identical to the ones exposed in valkeymodule.h. */
typedef ValkeyModuleString * (*ValkeyModuleConfigGetStringFunc)(const char *name, void *privdata);
typedef long long (*ValkeyModuleConfigGetNumericFunc)(const char *name, void *privdata);
typedef int (*ValkeyModuleConfigGetBoolFunc)(const char *name, void *privdata);
typedef int (*ValkeyModuleConfigGetEnumFunc)(const char *name, void *privdata);
/* The function signatures for module config set callbacks. These are identical to the ones exposed in valkeymodule.h. */
typedef int (*ValkeyModuleConfigSetStringFunc)(const char *name, ValkeyModuleString *val, void *privdata, ValkeyModuleString **err);
typedef int (*ValkeyModuleConfigSetNumericFunc)(const char *name, long long val, void *privdata, ValkeyModuleString **err);
typedef int (*ValkeyModuleConfigSetBoolFunc)(const char *name, int val, void *privdata, ValkeyModuleString **err);
typedef int (*ValkeyModuleConfigSetEnumFunc)(const char *name, int val, void *privdata, ValkeyModuleString **err);
/* Apply signature, identical to valkeymodule.h */
typedef int (*ValkeyModuleConfigApplyFunc)(ValkeyModuleCtx *ctx, void *privdata, ValkeyModuleString **err);
/* Struct representing a module config. These are stored in a list in the module struct */
struct ModuleConfig {
sds name; /* Name of config without the module name appended to the front */
void *privdata; /* Optional data passed into the module config callbacks */
union get_fn { /* The get callback specified by the module */
ValkeyModuleConfigGetStringFunc get_string;
ValkeyModuleConfigGetNumericFunc get_numeric;
ValkeyModuleConfigGetBoolFunc get_bool;
ValkeyModuleConfigGetEnumFunc get_enum;
} get_fn;
union set_fn { /* The set callback specified by the module */
ValkeyModuleConfigSetStringFunc set_string;
ValkeyModuleConfigSetNumericFunc set_numeric;
ValkeyModuleConfigSetBoolFunc set_bool;
ValkeyModuleConfigSetEnumFunc set_enum;
} set_fn;
ValkeyModuleConfigApplyFunc apply_fn;
ValkeyModule *module;
};
typedef struct ValkeyModuleAsyncRMCallPromise{
size_t ref_count;
void *private_data;
ValkeyModule *module;
ValkeyModuleOnUnblocked on_unblocked;
client *c;
ValkeyModuleCtx *ctx;
} ValkeyModuleAsyncRMCallPromise;
/* --------------------------------------------------------------------------
* Prototypes
* -------------------------------------------------------------------------- */
void VM_FreeCallReply(ValkeyModuleCallReply *reply);
void VM_CloseKey(ValkeyModuleKey *key);
void autoMemoryCollect(ValkeyModuleCtx *ctx);
robj **moduleCreateArgvFromUserFormat(const char *cmdname, const char *fmt, int *argcp, int *flags, va_list ap);
void VM_ZsetRangeStop(ValkeyModuleKey *kp);
static void zsetKeyReset(ValkeyModuleKey *key);
static void moduleInitKeyTypeSpecific(ValkeyModuleKey *key);
void VM_FreeDict(ValkeyModuleCtx *ctx, ValkeyModuleDict *d);
void VM_FreeServerInfo(ValkeyModuleCtx *ctx, ValkeyModuleServerInfoData *data);
/* Helpers for VM_SetCommandInfo. */
static int moduleValidateCommandInfo(const ValkeyModuleCommandInfo *info);
static int64_t moduleConvertKeySpecsFlags(int64_t flags, int from_api);
static int moduleValidateCommandArgs(ValkeyModuleCommandArg *args,
const ValkeyModuleCommandInfoVersion *version);
static struct serverCommandArg *moduleCopyCommandArgs(ValkeyModuleCommandArg *args,
const ValkeyModuleCommandInfoVersion *version);
static serverCommandArgType moduleConvertArgType(ValkeyModuleCommandArgType type, int *error);
static int moduleConvertArgFlags(int flags);
void moduleCreateContext(ValkeyModuleCtx *out_ctx, ValkeyModule *module, int ctx_flags);
/* Common helper functions. */
int moduleVerifyResourceName(const char *name);
/* --------------------------------------------------------------------------
* ## Heap allocation raw functions
*
* Memory allocated with these functions are taken into account by key
* eviction algorithms and are reported in memory usage information.
* -------------------------------------------------------------------------- */
/* Use like malloc(). Memory allocated with this function is reported in
* INFO memory, used for keys eviction according to maxmemory settings
* and in general is taken into account as memory allocated by the server.
* You should avoid using malloc().
* This function panics if unable to allocate enough memory. */
void *VM_Alloc(size_t bytes) {
/* Use 'zmalloc_usable()' instead of 'zmalloc()' to allow the compiler
* to recognize the additional memory size, which means that modules can
* use the memory reported by 'VM_MallocUsableSize()' safely. In theory this
* isn't really needed since this API can't be inlined (not even for embedded
* modules like TLS (we use function pointers for module APIs), and the API doesn't
* have the malloc_size attribute, but it's hard to predict how smart future compilers
* will be, so better safe than sorry. */
return zmalloc_usable(bytes,NULL);
}
/* Similar to VM_Alloc, but returns NULL in case of allocation failure, instead
* of panicking. */
void *VM_TryAlloc(size_t bytes) {
return ztrymalloc_usable(bytes,NULL);
}
/* Use like calloc(). Memory allocated with this function is reported in
* INFO memory, used for keys eviction according to maxmemory settings
* and in general is taken into account as memory allocated by the server.
* You should avoid using calloc() directly. */
void *VM_Calloc(size_t nmemb, size_t size) {
return zcalloc_usable(nmemb*size,NULL);
}
/* Similar to VM_Calloc, but returns NULL in case of allocation failure, instead
* of panicking. */
void *VM_TryCalloc(size_t nmemb, size_t size) {
return ztrycalloc_usable(nmemb*size,NULL);
}
/* Use like realloc() for memory obtained with ValkeyModule_Alloc(). */
void* VM_Realloc(void *ptr, size_t bytes) {
return zrealloc_usable(ptr,bytes,NULL);
}
/* Similar to VM_Realloc, but returns NULL in case of allocation failure,
* instead of panicking. */
void *VM_TryRealloc(void *ptr, size_t bytes) {
return ztryrealloc_usable(ptr,bytes,NULL);
}
/* Use like free() for memory obtained by ValkeyModule_Alloc() and
* ValkeyModule_Realloc(). However you should never try to free with
* ValkeyModule_Free() memory allocated with malloc() inside your module. */
void VM_Free(void *ptr) {
zfree(ptr);
}
/* Like strdup() but returns memory allocated with ValkeyModule_Alloc(). */
char *VM_Strdup(const char *str) {
return zstrdup(str);
}
/* --------------------------------------------------------------------------
* Pool allocator
* -------------------------------------------------------------------------- */
/* Release the chain of blocks used for pool allocations. */
void poolAllocRelease(ValkeyModuleCtx *ctx) {
ValkeyModulePoolAllocBlock *head = ctx->pa_head, *next;
while(head != NULL) {
next = head->next;
zfree(head);
head = next;
}
ctx->pa_head = NULL;
}
/* Return heap allocated memory that will be freed automatically when the
* module callback function returns. Mostly suitable for small allocations
* that are short living and must be released when the callback returns
* anyway. The returned memory is aligned to the architecture word size
* if at least word size bytes are requested, otherwise it is just
* aligned to the next power of two, so for example a 3 bytes request is
* 4 bytes aligned while a 2 bytes request is 2 bytes aligned.
*
* There is no realloc style function since when this is needed to use the
* pool allocator is not a good idea.
*
* The function returns NULL if `bytes` is 0. */
void *VM_PoolAlloc(ValkeyModuleCtx *ctx, size_t bytes) {
if (bytes == 0) return NULL;
ValkeyModulePoolAllocBlock *b = ctx->pa_head;
size_t left = b ? b->size - b->used : 0;
/* Fix alignment. */
if (left >= bytes) {
size_t alignment = VALKEYMODULE_POOL_ALLOC_ALIGN;
while (bytes < alignment && alignment/2 >= bytes) alignment /= 2;
if (b->used % alignment)
b->used += alignment - (b->used % alignment);
left = (b->used > b->size) ? 0 : b->size - b->used;
}
/* Create a new block if needed. */
if (left < bytes) {
size_t blocksize = VALKEYMODULE_POOL_ALLOC_MIN_SIZE;
if (blocksize < bytes) blocksize = bytes;
b = zmalloc(sizeof(*b) + blocksize);
b->size = blocksize;
b->used = 0;
b->next = ctx->pa_head;
ctx->pa_head = b;
}
char *retval = b->memory + b->used;
b->used += bytes;
return retval;
}
/* --------------------------------------------------------------------------
* Helpers for modules API implementation
* -------------------------------------------------------------------------- */
client *moduleAllocTempClient(void) {
client *c = NULL;
if (moduleTempClientCount > 0) {
c = moduleTempClients[--moduleTempClientCount];
if (moduleTempClientCount < moduleTempClientMinCount)
moduleTempClientMinCount = moduleTempClientCount;
} else {
c = createClient(NULL);
c->flags |= CLIENT_MODULE;
c->user = NULL; /* Root user */
}
return c;
}
static void freeRedisModuleAsyncRMCallPromise(ValkeyModuleAsyncRMCallPromise *promise) {
if (--promise->ref_count > 0) {
return;
}
/* When the promise is finally freed it can not have a client attached to it.
* Either releasing the client or VM_CallReplyPromiseAbort would have removed it. */
serverAssert(!promise->c);
zfree(promise);
}
void moduleReleaseTempClient(client *c) {
if (moduleTempClientCount == moduleTempClientCap) {
moduleTempClientCap = moduleTempClientCap ? moduleTempClientCap*2 : 32;
moduleTempClients = zrealloc(moduleTempClients, sizeof(c)*moduleTempClientCap);
}
clearClientConnectionState(c);
listEmpty(c->reply);
c->reply_bytes = 0;
c->duration = 0;
resetClient(c);
c->bufpos = 0;
c->flags = CLIENT_MODULE;
c->user = NULL; /* Root user */
c->cmd = c->lastcmd = c->realcmd = NULL;
if (c->bstate.async_rm_call_handle) {
ValkeyModuleAsyncRMCallPromise *promise = c->bstate.async_rm_call_handle;
promise->c = NULL; /* Remove the client from the promise so it will no longer be possible to abort it. */
freeRedisModuleAsyncRMCallPromise(promise);
c->bstate.async_rm_call_handle = NULL;
}
moduleTempClients[moduleTempClientCount++] = c;
}
/* Create an empty key of the specified type. `key` must point to a key object
* opened for writing where the `.value` member is set to NULL because the
* key was found to be non existing.
*
* On success VALKEYMODULE_OK is returned and the key is populated with
* the value of the specified type. The function fails and returns
* VALKEYMODULE_ERR if:
*
* 1. The key is not open for writing.
* 2. The key is not empty.
* 3. The specified type is unknown.
*/
int moduleCreateEmptyKey(ValkeyModuleKey *key, int type) {
robj *obj;
/* The key must be open for writing and non existing to proceed. */
if (!(key->mode & VALKEYMODULE_WRITE) || key->value)
return VALKEYMODULE_ERR;
switch(type) {
case VALKEYMODULE_KEYTYPE_LIST:
obj = createListListpackObject();
break;
case VALKEYMODULE_KEYTYPE_ZSET:
obj = createZsetListpackObject();
break;
case VALKEYMODULE_KEYTYPE_HASH:
obj = createHashObject();
break;
case VALKEYMODULE_KEYTYPE_STREAM:
obj = createStreamObject();
break;
default: return VALKEYMODULE_ERR;
}
dbAdd(key->db,key->key,obj);
key->value = obj;
moduleInitKeyTypeSpecific(key);
return VALKEYMODULE_OK;
}
/* Frees key->iter and sets it to NULL. */
static void moduleFreeKeyIterator(ValkeyModuleKey *key) {
serverAssert(key->iter != NULL);
switch (key->value->type) {
case OBJ_LIST: listTypeReleaseIterator(key->iter); break;
case OBJ_STREAM:
streamIteratorStop(key->iter);
zfree(key->iter);
break;
default: serverAssert(0); /* No key->iter for other types. */
}
key->iter = NULL;
}
/* Callback for listTypeTryConversion().
* Frees list iterator and sets it to NULL. */
static void moduleFreeListIterator(void *data) {
ValkeyModuleKey *key = (ValkeyModuleKey*)data;
serverAssert(key->value->type == OBJ_LIST);
if (key->iter) moduleFreeKeyIterator(key);
}
/* This function is called in low-level API implementation functions in order
* to check if the value associated with the key remained empty after an
* operation that removed elements from an aggregate data type.
*
* If this happens, the key is deleted from the DB and the key object state
* is set to the right one in order to be targeted again by write operations
* possibly recreating the key if needed.
*
* The function returns 1 if the key value object is found empty and is
* deleted, otherwise 0 is returned. */
int moduleDelKeyIfEmpty(ValkeyModuleKey *key) {
if (!(key->mode & VALKEYMODULE_WRITE) || key->value == NULL) return 0;
int isempty;
robj *o = key->value;
switch(o->type) {
case OBJ_LIST: isempty = listTypeLength(o) == 0; break;
case OBJ_SET: isempty = setTypeSize(o) == 0; break;
case OBJ_ZSET: isempty = zsetLength(o) == 0; break;
case OBJ_HASH: isempty = hashTypeLength(o) == 0; break;
case OBJ_STREAM: isempty = streamLength(o) == 0; break;
default: isempty = 0;
}
if (isempty) {
if (key->iter) moduleFreeKeyIterator(key);
dbDelete(key->db,key->key);
key->value = NULL;
return 1;
} else {
return 0;
}
}
/* --------------------------------------------------------------------------
* Service API exported to modules
*
* Note that all the exported APIs are called VM_<funcname> in the core
* and ValkeyModule_<funcname> in the module side (defined as function
* pointers in valkeymodule.h). In this way the dynamic linker does not
* mess with our global function pointers, overriding it with the symbols
* defined in the main executable having the same names.
* -------------------------------------------------------------------------- */
int VM_GetApi(const char *funcname, void **targetPtrPtr) {
/* Lookup the requested module API and store the function pointer into the
* target pointer. The function returns VALKEYMODULE_ERR if there is no such
* named API, otherwise VALKEYMODULE_OK.
*
* This function is not meant to be used by modules developer, it is only
* used implicitly by including valkeymodule.h. */
dictEntry *he = dictFind(server.moduleapi, funcname);
if (!he) return VALKEYMODULE_ERR;
*targetPtrPtr = dictGetVal(he);
return VALKEYMODULE_OK;
}
void modulePostExecutionUnitOperations(void) {
if (server.execution_nesting)
return;
if (server.busy_module_yield_flags) {
blockingOperationEnds();
server.busy_module_yield_flags = BUSY_MODULE_YIELD_NONE;
if (server.current_client)
unprotectClient(server.current_client);
unblockPostponedClients();
}
}
/* Free the context after the user function was called. */
void moduleFreeContext(ValkeyModuleCtx *ctx) {
/* See comment in moduleCreateContext */
if (!(ctx->flags & (VALKEYMODULE_CTX_THREAD_SAFE|VALKEYMODULE_CTX_COMMAND))) {
exitExecutionUnit();
postExecutionUnitOperations();
}
autoMemoryCollect(ctx);
poolAllocRelease(ctx);
if (ctx->postponed_arrays) {
zfree(ctx->postponed_arrays);
ctx->postponed_arrays_count = 0;
serverLog(LL_WARNING,
"API misuse detected in module %s: "
"ValkeyModule_ReplyWith*(VALKEYMODULE_POSTPONED_LEN) "
"not matched by the same number of ValkeyModule_SetReply*Len() "
"calls.",
ctx->module->name);
}
/* If this context has a temp client, we return it back to the pool.
* If this context created a new client (e.g detached context), we free it.
* If the client is assigned manually, e.g ctx->client = someClientInstance,
* none of these flags will be set and we do not attempt to free it. */
if (ctx->flags & VALKEYMODULE_CTX_TEMP_CLIENT)
moduleReleaseTempClient(ctx->client);
else if (ctx->flags & VALKEYMODULE_CTX_NEW_CLIENT)
freeClient(ctx->client);
}
static CallReply *moduleParseReply(client *c, ValkeyModuleCtx *ctx) {
/* Convert the result of the command into a module reply. */
sds proto = sdsnewlen(c->buf,c->bufpos);
c->bufpos = 0;
while(listLength(c->reply)) {
clientReplyBlock *o = listNodeValue(listFirst(c->reply));
proto = sdscatlen(proto,o->buf,o->used);
listDelNode(c->reply,listFirst(c->reply));
}
CallReply *reply = callReplyCreate(proto, c->deferred_reply_errors, ctx);
c->deferred_reply_errors = NULL; /* now the responsibility of the reply object. */
return reply;
}
void moduleCallCommandUnblockedHandler(client *c) {
ValkeyModuleCtx ctx;
ValkeyModuleAsyncRMCallPromise *promise = c->bstate.async_rm_call_handle;
serverAssert(promise);
ValkeyModule *module = promise->module;
if (!promise->on_unblocked) {
moduleReleaseTempClient(c);
return; /* module did not set any unblock callback. */
}
moduleCreateContext(&ctx, module, VALKEYMODULE_CTX_TEMP_CLIENT);
selectDb(ctx.client, c->db->id);
CallReply *reply = moduleParseReply(c, NULL);
module->in_call++;
promise->on_unblocked(&ctx, reply, promise->private_data);
module->in_call--;
moduleFreeContext(&ctx);
moduleReleaseTempClient(c);
}
/* Create a module ctx and keep track of the nesting level.
*
* Note: When creating ctx for threads (VM_GetThreadSafeContext and
* VM_GetDetachedThreadSafeContext) we do not bump up the nesting level
* because we only need to track of nesting level in the main thread
* (only the main thread uses propagatePendingCommands) */
void moduleCreateContext(ValkeyModuleCtx *out_ctx, ValkeyModule *module, int ctx_flags) {
memset(out_ctx, 0 ,sizeof(ValkeyModuleCtx));
out_ctx->getapifuncptr = (void*)(unsigned long)&VM_GetApi;
out_ctx->module = module;
out_ctx->flags = ctx_flags;
if (ctx_flags & VALKEYMODULE_CTX_TEMP_CLIENT)
out_ctx->client = moduleAllocTempClient();
else if (ctx_flags & VALKEYMODULE_CTX_NEW_CLIENT)
out_ctx->client = createClient(NULL);
/* Calculate the initial yield time for long blocked contexts.
* in loading we depend on the server hz, but in other cases we also wait
* for busy_reply_threshold.
* Note that in theory we could have started processing BUSY_MODULE_YIELD_EVENTS
* sooner, and only delay the processing for clients till the busy_reply_threshold,
* but this carries some overheads of frequently marking clients with BLOCKED_POSTPONE
* and releasing them, i.e. if modules only block for short periods. */
if (server.loading)
out_ctx->next_yield_time = getMonotonicUs() + 1000000 / server.hz;
else
out_ctx->next_yield_time = getMonotonicUs() + server.busy_reply_threshold * 1000;
/* Increment the execution_nesting counter (module is about to execute some code),
* except in the following cases:
* 1. We came here from cmd->proc (either call() or AOF load).
* In the former, the counter has been already incremented from within
* call() and in the latter we don't care about execution_nesting
* 2. If we are running in a thread (execution_nesting will be dealt with
* when locking/unlocking the GIL) */
if (!(ctx_flags & (VALKEYMODULE_CTX_THREAD_SAFE|VALKEYMODULE_CTX_COMMAND))) {
enterExecutionUnit(1, 0);
}
}
/* This command binds the normal command invocation with commands
* exported by modules. */
void ValkeyModuleCommandDispatcher(client *c) {
ValkeyModuleCommand *cp = c->cmd->module_cmd;
ValkeyModuleCtx ctx;
moduleCreateContext(&ctx, cp->module, VALKEYMODULE_CTX_COMMAND);
ctx.client = c;
cp->func(&ctx,(void**)c->argv,c->argc);
moduleFreeContext(&ctx);
/* In some cases processMultibulkBuffer uses sdsMakeRoomFor to
* expand the query buffer, and in order to avoid a big object copy
* the query buffer SDS may be used directly as the SDS string backing
* the client argument vectors: sometimes this will result in the SDS
* string having unused space at the end. Later if a module takes ownership
* of the RedisString, such space will be wasted forever. Inside the
* server core this is not a problem because tryObjectEncoding() is called
* before storing strings in the key space. Here we need to do it
* for the module. */
for (int i = 0; i < c->argc; i++) {
/* Only do the work if the module took ownership of the object:
* in that case the refcount is no longer 1. */
if (c->argv[i]->refcount > 1)
trimStringObjectIfNeeded(c->argv[i], 0);
}
}
/* This function returns the list of keys, with the same interface as the
* 'getkeys' function of the native commands, for module commands that exported
* the "getkeys-api" flag during the registration. This is done when the
* list of keys are not at fixed positions, so that first/last/step cannot
* be used.
*
* In order to accomplish its work, the module command is called, flagging
* the context in a way that the command can recognize this is a special
* "get keys" call by calling ValkeyModule_IsKeysPositionRequest(ctx). */
int moduleGetCommandKeysViaAPI(struct serverCommand *cmd, robj **argv, int argc, getKeysResult *result) {
ValkeyModuleCommand *cp = cmd->module_cmd;
ValkeyModuleCtx ctx;
moduleCreateContext(&ctx, cp->module, VALKEYMODULE_CTX_KEYS_POS_REQUEST);
/* Initialize getKeysResult */
getKeysPrepareResult(result, MAX_KEYS_BUFFER);
ctx.keys_result = result;
cp->func(&ctx,(void**)argv,argc);
/* We currently always use the array allocated by VM_KeyAtPos() and don't try
* to optimize for the pre-allocated buffer.
*/
moduleFreeContext(&ctx);
return result->numkeys;
}
/* This function returns the list of channels, with the same interface as
* moduleGetCommandKeysViaAPI, for modules that declare "getchannels-api"
* during registration. Unlike keys, this is the only way to declare channels. */
int moduleGetCommandChannelsViaAPI(struct serverCommand *cmd, robj **argv, int argc, getKeysResult *result) {
ValkeyModuleCommand *cp = cmd->module_cmd;
ValkeyModuleCtx ctx;
moduleCreateContext(&ctx, cp->module, VALKEYMODULE_CTX_CHANNELS_POS_REQUEST);
/* Initialize getKeysResult */
getKeysPrepareResult(result, MAX_KEYS_BUFFER);
ctx.keys_result = result;
cp->func(&ctx,(void**)argv,argc);
/* We currently always use the array allocated by VM_RM_ChannelAtPosWithFlags() and don't try
* to optimize for the pre-allocated buffer. */
moduleFreeContext(&ctx);
return result->numkeys;
}
/* --------------------------------------------------------------------------
* ## Commands API