Skip to content

Commit ceef96d

Browse files
pajgopajgo
committed
Sample thread pool
1 parent 85c2729 commit ceef96d

File tree

9 files changed

+510
-29
lines changed

9 files changed

+510
-29
lines changed

‎csrc/redis-filtered-sort/client_list.c‎

Lines changed: 97 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,97 @@
1+
#include "client_list.h"
2+
3+
blocked_client_t *blocked_client_create(RedisModuleBlockedClient*bc, void **targ) {
4+
blocked_client_t *bct = malloc(sizeof(blocked_client_t));
5+
bct->client = bc;
6+
bct->targ = targ;
7+
return bct;
8+
}
9+
10+
int blocked_client_free(blocked_client_t *bct) {
11+
free(&bct->targ);
12+
free(bct);
13+
return 1;
14+
}
15+
16+
//Blocked clients array
17+
blocked_clients_t *blocked_clients_create(const char * cmd) {
18+
blocked_clients_t* blocked = malloc(sizeof(blocked_clients_t));
19+
blocked->clients = malloc(sizeof(RedisModuleBlockedClient*));
20+
blocked->size = 0;
21+
blocked->cmd = strdup(cmd);
22+
return blocked;
23+
}
24+
25+
int blocked_clients_add(blocked_clients_t *bc, RedisModuleBlockedClient *client, void **targ) {
26+
bc->clients = realloc(bc->clients, sizeof(blocked_client_t*) * (bc->size +1));
27+
bc->clients[bc->size] = blocked_client_create(client, targ);
28+
bc->size++;
29+
return 1;
30+
}
31+
32+
int blocked_clients_free(blocked_clients_t *bc) {
33+
for (size_t i = 0; i < bc->size; i++){
34+
blocked_client_free(bc->clients[i]);
35+
}
36+
free(bc->clients);
37+
free(bc);
38+
return 1;
39+
}
40+
41+
//General blocklist maps blocked clients and command
42+
blocked_list_t *blocked_list_create(const char * cmd) {
43+
blocked_list_t* bl = malloc(sizeof(blocked_list_t));
44+
bl->list = malloc(sizeof(blocked_clients_t*));
45+
bl->size = 0;
46+
bl->cmd = strdup(cmd);
47+
return bl;
48+
}
49+
50+
int blocked_list_add_bc(blocked_list_t *bl, blocked_clients_t *clientList) {
51+
bl->list = realloc(bl->list, sizeof(blocked_clients_t*) * (bl->size +1));
52+
bl->list[bl->size] = clientList;
53+
bl->size++;
54+
return 1;
55+
}
56+
57+
int blocked_list_delete(blocked_list_t *bl, blocked_clients_t *clientList) {
58+
int pos = -1;
59+
for (size_t i = 0; i < bl->size; i++){
60+
if (bl->list[i] == clientList) pos = i;
61+
}
62+
63+
if (pos == -1) return 0;
64+
65+
blocked_clients_free(clientList);
66+
67+
for (size_t i = pos - 1; 0 < bl->size - 1; i++)
68+
bl->list[i] = bl->list[i+1];
69+
bl->size --;
70+
bl->list = realloc(bl->list, sizeof(blocked_clients_t*));
71+
return 1;
72+
}
73+
74+
//Searches clients blocked by command
75+
blocked_clients_t *blocked_list_find_cmd(blocked_list_t *list, const char *cmd) {
76+
for (size_t i = 0; i < list->size; i++) {
77+
if (strcasecmp(list->list[i]->cmd, cmd)) {
78+
return list->list[i];
79+
}
80+
}
81+
return NULL;
82+
}
83+
84+
//Add blocked client into specified group
85+
int blocked_list_add_client(blocked_list_t *list, const char *cmd, RedisModuleBlockedClient *client, void **targ) {
86+
blocked_clients_t *bc;
87+
bc = blocked_list_find_cmd(list, cmd);
88+
//create new bock list if not exists
89+
if (bc == NULL) {
90+
bc = blocked_clients_create(cmd);
91+
blocked_list_add_bc(list, bc);
92+
}
93+
blocked_clients_add(bc, client, targ);
94+
return 1;
95+
}
96+
97+

‎csrc/redis-filtered-sort/client_list.h‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
#ifndef __CLIENTS_LIST_H
2+
#define __CLIENTS_LIST_H
3+
4+
#include "general.h"
5+
6+
typedef struct blocked_client {
7+
RedisModuleBlockedClient *client;
8+
void ** targ;
9+
} blocked_client_t;
10+
11+
typedef struct blocked_clients {
12+
blocked_client_t **clients;
13+
size_t size;
14+
const char * cmd;
15+
} blocked_clients_t;
16+
17+
typedef struct blocked_list {
18+
blocked_clients_t **list;
19+
size_t size;
20+
const char * cmd;
21+
} blocked_list_t;
22+
23+
blocked_clients_t *blocked_list_find_cmd(blocked_list_t *list, const char *cmd);
24+
int blocked_list_add_client(blocked_list_t *list, const char *cmd, RedisModuleBlockedClient *client, void **targ);
25+
blocked_list_t *blocked_list_create(const char * cmd);
26+
int blocked_client_free(blocked_client_t *bct);
27+
int blocked_clients_free(blocked_clients_t *bc);
28+
int blocked_list_delete(blocked_list_t *bl, blocked_clients_t *clientList);
29+
30+
#endif

‎csrc/redis-filtered-sort/filter_module.c‎

Lines changed: 103 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,12 @@
11
#include "filter_module.h"
22
#include "fsort.h"
33
#include "fsort_utils.h"
4+
#include "thread_pool.h"
5+
#include "client_list.h"
6+
7+
static FSortPool_t *sortPool;
8+
9+
static blocked_list_t *fsortBlocked;
410

511
int FSortBust_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
612
RedisModule_AutoMemory(ctx);
@@ -26,53 +32,129 @@ int FSortBust_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int ar
2632
}
2733

2834
int FSortAggregate_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
35+
const char *cmd;
2936
RedisModule_AutoMemory(ctx);
3037

3138
if (argc != 4) {
3239
return RedisModule_WrongArity(ctx);
3340
}
3441

35-
pthread_t tid;
42+
cmd = strdup(GetArgvString(ctx, argv, argc));
43+
3644
RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx,NULL,NULL,NULL,0);
3745

3846
void **targ = RedisModule_Alloc(sizeof(void*)*3);
3947
targ[0] = bc;
4048
targ[1] = (void*)(unsigned long) argc;
4149
targ[2] = (void*)(RedisModuleString **)argv;
4250

43-
if (pthread_create(&tid,NULL,fsort_aggregate_thread,targ) != 0) {
44-
RedisModule_AbortBlock(bc);
45-
return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread");
51+
//run threads
52+
if (tpool_work_exists(sortPool, cmd) == 1) {
53+
blocked_list_add_client(fsortBlocked, cmd, bc, targ);
54+
} else {
55+
tpool_add_work(sortPool, fsort_aggregate_thread, targ, cmd);
4656
}
47-
57+
free((char*)cmd);
58+
4859
return REDISMODULE_OK;
4960
}
5061

5162
int FSort_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
63+
const char * cmd;
64+
int pssType = 0;
65+
int fflType = 0;
66+
int longRunning =0;
67+
pthread_t tid;
68+
5269
RedisModule_AutoMemory(ctx);
5370

5471
if (argc < 7 ) {
5572
return RedisModule_WrongArity(ctx);
5673
}
5774

58-
pthread_t tid;
5975
RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx,NULL,NULL,NULL,0);
6076

61-
void **targ = RedisModule_Alloc(sizeof(void*)*3);
62-
targ[0] = bc;
63-
targ[1] = (void*)(unsigned long) argc;
64-
targ[2] = (void*)(RedisModuleString **)argv;
77+
FSortObj_t *sort = fsort_new_fsort();
78+
RedisModuleCtx *thSafeCtx = RedisModule_GetThreadSafeContext(bc);
79+
sort->ctx = ctx;
80+
int parseRes = fsort_parse_args(sort, thSafeCtx, argv, argc);
6581

66-
if (pthread_create(&tid,NULL,fsort_fsort_thread,targ) != 0) {
67-
RedisModule_AbortBlock(bc);
68-
return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread");
82+
fsort_form_keys(sort);
83+
84+
if (parseRes != REDISMODULE_OK) {
85+
fsort_free_fsort(sort);
86+
RedisModule_UnblockClient(bc,NULL);
87+
RedisModule_FreeThreadSafeContext(thSafeCtx);
88+
return REDISMODULE_ERR;
6989
}
7090

91+
pssType = fsort_key_type(sort, sort->pssKey);
92+
fflType = fsort_key_type(sort, sort->fflKey);
93+
94+
if (pssType == REDISMODULE_KEYTYPE_EMPTY || fflType == REDISMODULE_KEYTYPE_EMPTY) {
95+
longRunning = 1;
96+
//decide on which tasks block
97+
if (RedisModule_StringCompare(sort->pssKey, sort->fflKey) == 0) {
98+
//only sort, setting command to psskey
99+
cmd = RedisModule_StringPtrLen(sort->pssKey, NULL);
100+
} else {
101+
//if no PSS key block all commands
102+
//if no ffl block only filtering
103+
//only elseif
104+
if (pssType == REDISMODULE_KEYTYPE_EMPTY) {
105+
cmd = RedisModule_StringPtrLen(sort->pssKey, NULL);
106+
} else if (fflType == REDISMODULE_KEYTYPE_EMPTY) {
107+
cmd = RedisModule_StringPtrLen(sort->fflKey, NULL);
108+
}
109+
}
110+
cmd = strdup(cmd);
111+
}
112+
113+
void **targ = RedisModule_Alloc(sizeof(void*)*3);
114+
targ[0] = bc;
115+
targ[1] = (void *)sort;
116+
//if all keys are present, we dont need thread pool
117+
if (longRunning == 1) {
118+
//run threads
119+
if (tpool_work_exists(sortPool, cmd) == 1) {
120+
blocked_list_add_client(fsortBlocked, cmd, bc, targ);
121+
} else {
122+
tpool_add_work(sortPool, fsort_fsort_thread, targ, cmd);
123+
}
124+
free((char*)cmd);
125+
} else {
126+
if (pthread_create(&tid,NULL,fsort_fsort_thread,targ) != 0) {
127+
RedisModule_AbortBlock(bc);
128+
RedisModule_FreeThreadSafeContext(thSafeCtx);
129+
free(targ);
130+
return RedisModule_ReplyWithError(ctx,"-ERR Can't start thread");
131+
}
132+
}
133+
RedisModule_FreeThreadSafeContext(thSafeCtx);
71134
return REDISMODULE_OK;
72135
}
73136

74-
int RedisModule_OnLoad(RedisModuleCtx *ctx) {
137+
int FSort_onWorkFinish(FSortWork_t * work) {
138+
printf("[rm]WorkFinish %s\n", work->cmd);
139+
blocked_clients_t* clientList = blocked_list_find_cmd(fsortBlocked, work->cmd);
140+
141+
if(clientList != NULL) {
142+
for (size_t i = 0; i < clientList->size; i++) {
143+
blocked_client_t *bct = clientList->clients[i];
144+
tpool_add_work(sortPool, fsort_fsort_thread, bct->targ, work->cmd);
145+
}
146+
blocked_list_delete(fsortBlocked, clientList);
147+
}
75148

149+
free(work->argv);
150+
tpool_work_destroy(work);
151+
152+
printf("[rm]Unblock done!\n");
153+
return 1;
154+
}
155+
156+
int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
157+
long long poolSize = 2;
76158
if (RedisModule_Init(ctx, "FilterSortModule", 1, REDISMODULE_APIVER_1) == REDISMODULE_ERR) {
77159
return REDISMODULE_ERR;
78160
}
@@ -88,6 +170,13 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx) {
88170
if (RedisModule_CreateCommand(ctx, "fsortaggregate", FSortAggregate_RedisCommand, "write", 1, 2, 1) == REDISMODULE_ERR) {
89171
return REDISMODULE_ERR;
90172
}
173+
174+
if (argc == 1) {
175+
RedisModule_StringToLongLong(argv[0], &poolSize);
176+
}
91177

178+
sortPool = tpool_create(poolSize);
179+
sortPool->cb = FSort_onWorkFinish;
180+
fsortBlocked = blocked_list_create("fsort-f");
92181
return REDISMODULE_OK;
93182
}

‎csrc/redis-filtered-sort/fsort.c‎

Lines changed: 11 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -478,6 +478,13 @@ const char **fsort_filter_data(FSortObj_t *sort, size_t *outputDataSize) {
478478
return outputData;
479479
}
480480

481+
int fsort_key_type(FSortObj_t *sort, RedisModuleString *keyName) {
482+
RedisModuleKey *key = RedisModule_OpenKey(sort->ctx, keyName, REDISMODULE_READ);
483+
int kType = RedisModule_KeyType(key);
484+
RedisModule_CloseKey(key);
485+
return kType;
486+
}
487+
481488
int fsort_fsort(FSortObj_t *sort) {
482489
RedisModuleKey *fflkey = RedisModule_OpenKey(sort->ctx, sort->fflKey, REDISMODULE_READ);
483490
RedisModuleKey *psskey = RedisModule_OpenKey(sort->ctx, sort->pssKey, REDISMODULE_READ);
@@ -733,23 +740,13 @@ void *fsort_aggregate_thread(void *arg) {
733740
void *fsort_fsort_thread(void *arg) {
734741
void **targ = arg;
735742
RedisModuleBlockedClient *bc = targ[0];
736-
RedisModuleCtx *ctx = RedisModule_GetThreadSafeContext(bc);
737-
int argc = (unsigned long) targ[1];
738-
RedisModuleString **argv = targ[2];
739-
740-
FSortObj_t *sort = fsort_new_fsort();
741-
int parseRes = fsort_parse_args(sort, ctx, argv, argc);
742-
fsort_form_keys(sort);
743-
744-
if (parseRes == REDISMODULE_OK) {
745-
//FSortObj_t sobj = *sort;
746-
fsort_fsort(sort);
747-
}
748-
743+
FSortObj_t *sort = targ[1];
744+
RedisModuleCtx* ctx = RedisModule_GetThreadSafeContext(bc);
745+
sort->ctx = ctx;
746+
fsort_fsort(sort);
749747
fsort_free_fsort(sort);
750748

751749
RedisModule_UnblockClient(bc, NULL);
752-
753750
RedisModule_FreeThreadSafeContext(ctx);
754751

755752
return NULL;

‎csrc/redis-filtered-sort/fsort.h‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,5 +26,9 @@ int fsort_parse_args(FSortObj_t *sObj, RedisModuleCtx *ctx, RedisModuleString **
2626
FSortObj_t *fsort_new_fsort();
2727

2828
void fsort_cache_buster(FSortObj_t *sortObj, RedisModuleString *key);
29+
void fsort_free_fsort(FSortObj_t *sort);
30+
31+
int fsort_key_type(FSortObj_t *sort, RedisModuleString *keyName);
32+
2933

3034
#endif

0 commit comments

Comments
 (0)