lkn
/
pcloudcc-lneely
mirror of https://github.com/lneely/pcloudcc-lneely.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

remaining plist and ptask functions moved to appropriate namespaces

This commit is contained in:
Levi Neely 2025-03-14 21:04:53 +01:00
parent dc8ffdb1a4
commit ef38c69870
7 changed files with 407 additions and 416 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

401
pclsync/plibs.c
View File

@ -40,6 +40,7 @@
#include "pdbg.h"
#include "pfile.h"
#include "plibs.h"
#include "plist.h"
#include "plocks.h"
#include "pnetlibs.h"
#include "prun.h"
@ -471,6 +472,38 @@ void psync_sql_do_rdlock(const char *file, unsigned line) {
void psync_sql_rdlock() { plocks_rdlock(&psync_db_lock); }
#endif
void psync_list_bulder_add_sql(psync_list_builder_t *builder,
psync_sql_res *res,
psync_list_builder_sql_callback callback) {
psync_variant_row row;
while ((row = psync_sql_fetch_row(res))) {
if (!builder->last_elements ||
builder->last_elements->used >= builder->elements_per_list) {
builder->last_elements = (psync_list_element_list *)malloc(
offsetof(psync_list_element_list, elements) +
builder->element_size * builder->elements_per_list);
psync_list_add_tail(&builder->element_list,
&builder->last_elements->list);
builder->last_elements->used = 0;
}
builder->current_element =
builder->last_elements->elements +
builder->last_elements->used * builder->element_size;
builder->cstrcnt = psync_list_bulder_push_num(builder);
*builder->cstrcnt = 0;
while (callback(builder, builder->current_element, row)) {
row = psync_sql_fetch_row(res);
if (!row)
break;
*builder->cstrcnt = 0;
}
builder->last_elements->used++;
builder->cnt++;
}
psync_sql_free_result(res);
}
void psync_sql_rdunlock() {
#if IS_DEBUG
if (unlikely(sqlrdlockcnt == 0)) {
@ -1544,374 +1577,6 @@ int psync_match_pattern(const char *name, const char *pattern, size_t plen) {
return name[i] == 0;
}
typedef struct {
psync_list list;
unsigned long used;
char elements[];
} psync_list_element_list;
typedef struct {
psync_list list;
char *next;
char *end;
} psync_list_string_list;
typedef struct {
psync_list list;
unsigned long used;
uint32_t numbers[1000];
} psync_list_num_list;
struct psync_list_builder_t_ {
size_t element_size;
size_t elements_offset;
size_t elements_per_list;
size_t stringalloc;
uint64_t cnt;
psync_list element_list;
psync_list_element_list *last_elements;
psync_list string_list;
psync_list_string_list *last_strings;
psync_list number_list;
psync_list_num_list *last_numbers;
unsigned long popoff;
char *current_element;
uint32_t *cstrcnt;
};
psync_list_builder_t *psync_list_builder_create(size_t element_size,
size_t offset) {
psync_list_builder_t *builder;
builder = malloc(sizeof(psync_list_builder_t));
builder->element_size = element_size;
builder->elements_offset = offset;
if (element_size <= 200)
builder->elements_per_list = 40;
else
builder->elements_per_list = 12;
builder->cnt = 0;
builder->stringalloc = 0;
psync_list_init(&builder->element_list);
builder->last_elements = NULL;
psync_list_init(&builder->string_list);
builder->last_strings = NULL;
psync_list_init(&builder->number_list);
builder->last_numbers = NULL;
return builder;
}
static uint32_t *psync_list_bulder_push_num(psync_list_builder_t *builder) {
if (!builder->last_numbers ||
builder->last_numbers->used >=
sizeof(builder->last_numbers->numbers) / sizeof(uint32_t)) {
psync_list_num_list *l = malloc(sizeof(psync_list_num_list));
l->used = 0;
builder->last_numbers = l;
psync_list_add_tail(&builder->number_list, &l->list);
}
return &builder->last_numbers->numbers[builder->last_numbers->used++];
}
static uint32_t psync_list_bulder_pop_num(psync_list_builder_t *builder) {
uint32_t ret;
ret = builder->last_numbers->numbers[builder->popoff++];
if (builder->popoff >= builder->last_numbers->used) {
builder->last_numbers = psync_list_element(builder->last_numbers->list.next,
psync_list_num_list, list);
builder->popoff = 0;
}
return ret;
}
void psync_list_bulder_add_sql(psync_list_builder_t *builder,
psync_sql_res *res,
psync_list_builder_sql_callback callback) {
psync_variant_row row;
while ((row = psync_sql_fetch_row(res))) {
if (!builder->last_elements ||
builder->last_elements->used >= builder->elements_per_list) {
builder->last_elements = (psync_list_element_list *)malloc(
offsetof(psync_list_element_list, elements) +
builder->element_size * builder->elements_per_list);
psync_list_add_tail(&builder->element_list,
&builder->last_elements->list);
builder->last_elements->used = 0;
}
builder->current_element =
builder->last_elements->elements +
builder->last_elements->used * builder->element_size;
builder->cstrcnt = psync_list_bulder_push_num(builder);
*builder->cstrcnt = 0;
while (callback(builder, builder->current_element, row)) {
row = psync_sql_fetch_row(res);
if (!row)
break;
*builder->cstrcnt = 0;
}
builder->last_elements->used++;
builder->cnt++;
}
psync_sql_free_result(res);
}
void *psync_list_bulder_add_element(psync_list_builder_t *builder) {
if (!builder->last_elements ||
builder->last_elements->used >= builder->elements_per_list) {
builder->last_elements = (psync_list_element_list *)malloc(
offsetof(psync_list_element_list, elements) +
builder->element_size * builder->elements_per_list);
psync_list_add_tail(&builder->element_list, &builder->last_elements->list);
builder->last_elements->used = 0;
}
builder->current_element =
builder->last_elements->elements +
builder->last_elements->used * builder->element_size;
builder->cstrcnt = psync_list_bulder_push_num(builder);
*builder->cstrcnt = 0;
builder->last_elements->used++;
builder->cnt++;
return builder->current_element;
}
void psync_list_add_lstring_offset(psync_list_builder_t *builder, size_t offset,
size_t length) {
char **str, *s;
psync_list_string_list *l;
length++;
str = (char **)(builder->current_element + offset);
builder->stringalloc += length;
if (unlikely(length > 2000)) {
l = (psync_list_string_list *)malloc(sizeof(psync_list_string_list) +
length);
s = (char *)(l + 1);
psync_list_add_tail(&builder->string_list, &l->list);
} else if (!builder->last_strings || builder->last_strings->next + length >
builder->last_strings->end) {
l = (psync_list_string_list *)malloc(sizeof(psync_list_string_list) +
4000);
s = (char *)(l + 1);
l->next = s + length;
l->end = s + 4000;
psync_list_add_tail(&builder->string_list, &l->list);
builder->last_strings = l;
} else {
s = builder->last_strings->next;
builder->last_strings->next += length;
}
memcpy(s, *str, length);
*str = s;
*(psync_list_bulder_push_num(builder)) = offset;
*(psync_list_bulder_push_num(builder)) = length;
(*builder->cstrcnt)++;
}
void psync_list_add_string_offset(psync_list_builder_t *builder,
size_t offset) {
psync_list_add_lstring_offset(
builder, offset, strlen(*((char **)(builder->current_element + offset))));
}
void *psync_list_builder_finalize(psync_list_builder_t *builder) {
char *ret, *elem, *str;
char **pstr;
psync_list_element_list *el;
unsigned long i;
uint32_t j, scnt, offset, length;
size_t sz;
sz = builder->elements_offset + builder->element_size * builder->cnt +
builder->stringalloc;
pdbg_logf(D_NOTICE, "allocating %lu bytes, %lu of which for strings",
(unsigned long)sz, (unsigned long)builder->stringalloc);
ret = malloc(sizeof(char) * sz);
if (builder->elements_offset <= sizeof(builder->cnt))
memcpy(ret, &builder->cnt, builder->elements_offset);
else
memcpy(ret, &builder->cnt, sizeof(builder->cnt));
elem = ret + builder->elements_offset;
str = elem + builder->element_size * builder->cnt;
builder->last_numbers =
psync_list_element(builder->number_list.next, psync_list_num_list, list);
builder->popoff = 0;
psync_list_for_each_element(el, &builder->element_list,
psync_list_element_list, list) {
for (i = 0; i < el->used; i++) {
memcpy(elem, el->elements + (i * builder->element_size),
builder->element_size);
scnt = psync_list_bulder_pop_num(builder);
for (j = 0; j < scnt; j++) {
offset = psync_list_bulder_pop_num(builder);
length = psync_list_bulder_pop_num(builder);
pstr = (char **)(elem + offset);
memcpy(str, *pstr, length);
*pstr = str;
str += length;
}
elem += builder->element_size;
}
}
psync_list_for_each_element_call(&builder->element_list, psync_list_element_list, list, free);
psync_list_for_each_element_call(&builder->string_list, psync_list_string_list, list, free);
psync_list_for_each_element_call(&builder->number_list, psync_list_num_list, list, free);
free(builder);
return ret;
}
#define PSYNC_TASK_STATUS_RUNNING 0
#define PSYNC_TASK_STATUS_READY 1
#define PSYNC_TASK_STATUS_DONE 2
#define PSYNC_TASK_STATUS_RETURNED 3
struct psync_task_t_ {
psync_task_callback_t callback;
void *param;
pthread_cond_t cond;
int id;
int status;
};
// #define PSYNC_WAIT_ANYBODY -1 // unused, but may be important later
#define PSYNC_WAIT_NOBODY -2
#define PSYNC_WAIT_FREED -3
struct psync_task_manager_t_ {
pthread_mutex_t mutex;
int taskcnt;
int refcnt;
int waitfor;
struct psync_task_t_ tasks[];
};
static void psync_task_destroy(psync_task_manager_t tm) {
int i;
for (i = 0; i < tm->taskcnt; i++)
pthread_cond_destroy(&tm->tasks[i].cond);
pthread_mutex_destroy(&tm->mutex);
free(tm);
}
static void psync_task_dec_refcnt(psync_task_manager_t tm) {
int refcnt;
pthread_mutex_lock(&tm->mutex);
refcnt = --tm->refcnt;
pthread_mutex_unlock(&tm->mutex);
if (!refcnt)
psync_task_destroy(tm);
}
static psync_task_manager_t psync_get_manager_of_task(struct psync_task_t_ *t) {
return (psync_task_manager_t)(((char *)(t - t->id)) -
offsetof(struct psync_task_manager_t_, tasks));
}
static void psync_task_entry(void *ptr) {
struct psync_task_t_ *t;
t = (struct psync_task_t_ *)ptr;
t->callback(ptr, t->param);
psync_task_dec_refcnt(psync_get_manager_of_task(t));
}
psync_task_manager_t
psync_task_run_tasks(psync_task_callback_t const *callbacks,
void *const *params, int cnt) {
psync_task_manager_t ret;
struct psync_task_t_ *t;
int i;
ret = (psync_task_manager_t)malloc(
offsetof(struct psync_task_manager_t_, tasks) +
sizeof(struct psync_task_t_) * cnt);
pthread_mutex_init(&ret->mutex, NULL);
ret->taskcnt = cnt;
ret->refcnt = cnt + 1;
ret->waitfor = PSYNC_WAIT_NOBODY;
for (i = 0; i < cnt; i++) {
t = &ret->tasks[i];
t->callback = callbacks[i];
t->param = params[i];
pthread_cond_init(&t->cond, NULL);
t->id = i;
t->status = PSYNC_TASK_STATUS_RUNNING;
prun_thread1("task", psync_task_entry, t);
}
return ret;
}
void *psync_task_papi_result(psync_task_manager_t tm, int id) {
void *ret;
pthread_mutex_lock(&tm->mutex);
if (tm->tasks[id].status == PSYNC_TASK_STATUS_RUNNING) {
do {
tm->waitfor = id;
pthread_cond_wait(&tm->tasks[id].cond, &tm->mutex);
tm->waitfor = PSYNC_WAIT_NOBODY;
} while (tm->tasks[id].status == PSYNC_TASK_STATUS_RUNNING);
ret = tm->tasks[id].param;
tm->tasks[id].status = PSYNC_TASK_STATUS_DONE;
} else if (tm->tasks[id].status == PSYNC_TASK_STATUS_READY) {
ret = tm->tasks[id].param;
tm->tasks[id].status = PSYNC_TASK_STATUS_DONE;
pthread_cond_signal(&tm->tasks[id].cond);
} else {
pdbg_logf(D_BUG, "invalid status %d of task id %d", (int)tm->tasks[id].status,
id);
ret = NULL;
}
pthread_mutex_unlock(&tm->mutex);
return ret;
}
void psync_task_free(psync_task_manager_t tm) {
if (tm->refcnt == 1)
psync_task_destroy(tm);
else {
int refcnt, i;
pthread_mutex_lock(&tm->mutex);
tm->waitfor = PSYNC_WAIT_FREED;
for (i = 0; i < tm->taskcnt; i++)
if (tm->tasks[i].status == PSYNC_TASK_STATUS_READY) {
tm->tasks[i].status = PSYNC_TASK_STATUS_RETURNED;
pthread_cond_signal(&tm->tasks[i].cond);
}
refcnt = --tm->refcnt;
pthread_mutex_unlock(&tm->mutex);
if (!refcnt)
psync_task_destroy(tm);
}
}
int psync_task_complete(void *h, void *data) {
psync_task_manager_t tm;
struct psync_task_t_ *t;
int ret;
t = (struct psync_task_t_ *)h;
tm = psync_get_manager_of_task(t);
pthread_mutex_lock(&tm->mutex);
if (tm->waitfor == t->id) {
t->param = data;
t->status = PSYNC_TASK_STATUS_READY;
pthread_cond_signal(&t->cond);
ret = 0;
} else if (tm->waitfor == PSYNC_WAIT_NOBODY || tm->waitfor >= 0) {
t->param = data;
t->status = PSYNC_TASK_STATUS_READY;
do {
pthread_cond_wait(&t->cond, &tm->mutex);
} while (t->status == PSYNC_TASK_STATUS_READY);
if (t->status == PSYNC_TASK_STATUS_RETURNED)
ret = -1;
else
ret = 0;
} else if (tm->waitfor == PSYNC_WAIT_FREED)
ret = -1;
else {
pdbg_logf(D_BUG, "invalid waitfor value %d", tm->waitfor);
ret = -1;
}
pthread_mutex_unlock(&tm->mutex);
return ret;
}
static uint32_t pq_rnd() {
static uint32_t a = 0x95ae3d25, b = 0xe225d755, c = 0xc63a2ae7,

48
pclsync/plibs.h
View File

@ -167,24 +167,13 @@ typedef struct {
uint64_t data[];
} psync_full_result_int;
struct psync_list_builder_t_;
typedef struct psync_list_builder_t_ psync_list_builder_t;
struct psync_task_manager_t_;
typedef struct psync_task_manager_t_ *psync_task_manager_t;
typedef const uint64_t *psync_uint_row;
typedef const char *const *psync_str_row;
typedef const psync_variant *psync_variant_row;
typedef void (*psync_run_after_t)(void *);
typedef int (*psync_list_builder_sql_callback)(psync_list_builder_t *, void *,
psync_variant_row);
typedef void (*psync_task_callback_t)(void *, void *);
typedef int (*psync_list_builder_sql_callback)(psync_list_builder_t *, void *, psync_variant_row);
typedef void (*psync_transaction_callback_t)(void *);
extern int psync_do_run;
@ -309,36 +298,18 @@ void psync_list_add_lstring_offset(psync_list_builder_t *builder, size_t offset,
size_t length);
void *psync_list_builder_finalize(psync_list_builder_t *builder);
psync_task_manager_t
psync_task_run_tasks(psync_task_callback_t const *callbacks,
void *const *params, int cnt);
void *psync_task_papi_result(psync_task_manager_t tm, int id);
void psync_task_free(psync_task_manager_t tm);
int psync_task_complete(void *h, void *data);
void psync_pqsort(void *base, size_t cnt, size_t sort_first, size_t size,
int (*compar)(const void *, const void *));
void psync_qpartition(void *base, size_t cnt, size_t sort_first, size_t size,
int (*compar)(const void *, const void *));
void psync_pqsort(void *base, size_t cnt, size_t sort_first, size_t size, int (*compar)(const void *, const void *));
void psync_qpartition(void *base, size_t cnt, size_t sort_first, size_t size, int (*compar)(const void *, const void *));
void psync_try_free_memory();
uint64_t psync_err_number_expected(const char *file, const char *function,
int unsigned line,
const psync_variant *v) PSYNC_COLD;
const char *psync_err_string_expected(const char *file, const char *function,
int unsigned line,
const psync_variant *v) PSYNC_COLD;
const char *psync_lstring_expected(const char *file, const char *function,
int unsigned line, const psync_variant *v,
size_t *len) PSYNC_NONNULL(4, 5);
double psync_err_real_expected(const char *file, const char *function,
int unsigned line,
const psync_variant *v) PSYNC_COLD;
uint64_t psync_err_number_expected(const char *file, const char *function, int unsigned line, const psync_variant *v) PSYNC_COLD;
const char *psync_err_string_expected(const char *file, const char *function, int unsigned line, const psync_variant *v) PSYNC_COLD;
const char *psync_lstring_expected(const char *file, const char *function, int unsigned line, const psync_variant *v, size_t *len) PSYNC_NONNULL(4, 5);
double psync_err_real_expected(const char *file, const char *function, int unsigned line, const psync_variant *v) PSYNC_COLD;
/* needs 12 characters of buffer space on top of the length of the prefix */
static inline void psync_get_string_id(char *dst, const char *prefix,
uint64_t id) {
static inline void psync_get_string_id(char *dst, const char *prefix, uint64_t id) {
size_t plen;
plen = strlen(prefix);
dst = (char *)memcpy(dst, prefix, plen) + plen;
@ -350,8 +321,7 @@ static inline void psync_get_string_id(char *dst, const char *prefix,
}
/* needs 24 characters of buffer space on top of the length of the prefix */
static inline void psync_get_string_id2(char *dst, const char *prefix,
uint64_t id1, uint64_t id2) {
static inline void psync_get_string_id2(char *dst, const char *prefix, uint64_t id1, uint64_t id2) {
size_t plen;
plen = strlen(prefix);
dst = (char *)memcpy(dst, prefix, plen) + plen;

157
pclsync/plist.c
View File

@ -30,7 +30,10 @@
*/
#include "plist.h"
#include "pdbg.h"
#include "pfile.h"
#include <stdlib.h>
#include <string.h>
/* Fairly simple in-place merge sort with constant storage requirements.
*
@ -111,10 +114,7 @@ void psync_list_sort(psync_list *l, psync_list_compare cmp) {
l->prev = l1;
}
void psync_list_extract_repeating(psync_list *l1, psync_list *l2,
psync_list *extracted1,
psync_list *extracted2,
psync_list_compare cmp) {
void psync_list_extract_repeating(psync_list *l1, psync_list *l2, psync_list *extracted1, psync_list *extracted2, psync_list_compare cmp) {
psync_list *li1, *li2, *ln1, *ln2;
int cr;
psync_list_sort(l1, cmp);
@ -139,3 +139,152 @@ void psync_list_extract_repeating(psync_list *l1, psync_list *l2,
}
}
}
psync_list_builder_t *psync_list_builder_create(size_t element_size, size_t offset) {
psync_list_builder_t *builder;
builder = malloc(sizeof(psync_list_builder_t));
builder->element_size = element_size;
builder->elements_offset = offset;
if (element_size <= 200)
builder->elements_per_list = 40;
else
builder->elements_per_list = 12;
builder->cnt = 0;
builder->stringalloc = 0;
psync_list_init(&builder->element_list);
builder->last_elements = NULL;
psync_list_init(&builder->string_list);
builder->last_strings = NULL;
psync_list_init(&builder->number_list);
builder->last_numbers = NULL;
return builder;
}
uint32_t *psync_list_bulder_push_num(psync_list_builder_t *builder) {
if (!builder->last_numbers ||
builder->last_numbers->used >=
sizeof(builder->last_numbers->numbers) / sizeof(uint32_t)) {
psync_list_num_list *l = malloc(sizeof(psync_list_num_list));
l->used = 0;
builder->last_numbers = l;
psync_list_add_tail(&builder->number_list, &l->list);
}
return &builder->last_numbers->numbers[builder->last_numbers->used++];
}
uint32_t psync_list_bulder_pop_num(psync_list_builder_t *builder) {
uint32_t ret;
ret = builder->last_numbers->numbers[builder->popoff++];
if (builder->popoff >= builder->last_numbers->used) {
builder->last_numbers = psync_list_element(builder->last_numbers->list.next,
psync_list_num_list, list);
builder->popoff = 0;
}
return ret;
}
void *psync_list_bulder_add_element(psync_list_builder_t *builder) {
if (!builder->last_elements ||
builder->last_elements->used >= builder->elements_per_list) {
builder->last_elements = (psync_list_element_list *)malloc(
offsetof(psync_list_element_list, elements) +
builder->element_size * builder->elements_per_list);
psync_list_add_tail(&builder->element_list, &builder->last_elements->list);
builder->last_elements->used = 0;
}
builder->current_element =
builder->last_elements->elements +
builder->last_elements->used * builder->element_size;
builder->cstrcnt = psync_list_bulder_push_num(builder);
*builder->cstrcnt = 0;
builder->last_elements->used++;
builder->cnt++;
return builder->current_element;
}
void psync_list_add_lstring_offset(psync_list_builder_t *builder, size_t offset, size_t length) {
char **str, *s;
psync_list_string_list *l;
length++;
str = (char **)(builder->current_element + offset);
builder->stringalloc += length;
if (unlikely(length > 2000)) {
l = (psync_list_string_list *)malloc(sizeof(psync_list_string_list) +
length);
s = (char *)(l + 1);
psync_list_add_tail(&builder->string_list, &l->list);
} else if (!builder->last_strings || builder->last_strings->next + length >
builder->last_strings->end) {
l = (psync_list_string_list *)malloc(sizeof(psync_list_string_list) +
4000);
s = (char *)(l + 1);
l->next = s + length;
l->end = s + 4000;
psync_list_add_tail(&builder->string_list, &l->list);
builder->last_strings = l;
} else {
s = builder->last_strings->next;
builder->last_strings->next += length;
}
memcpy(s, *str, length);
*str = s;
*(psync_list_bulder_push_num(builder)) = offset;
*(psync_list_bulder_push_num(builder)) = length;
(*builder->cstrcnt)++;
}
void psync_list_add_string_offset(psync_list_builder_t *builder,
size_t offset) {
psync_list_add_lstring_offset(
builder, offset, strlen(*((char **)(builder->current_element + offset))));
}
void *psync_list_builder_finalize(psync_list_builder_t *builder) {
char *ret, *elem, *str;
char **pstr;
psync_list_element_list *el;
unsigned long i;
uint32_t j, scnt, offset, length;
size_t sz;
sz = builder->elements_offset + builder->element_size * builder->cnt +
builder->stringalloc;
pdbg_logf(D_NOTICE, "allocating %lu bytes, %lu of which for strings",
(unsigned long)sz, (unsigned long)builder->stringalloc);
ret = malloc(sizeof(char) * sz);
if (builder->elements_offset <= sizeof(builder->cnt))
memcpy(ret, &builder->cnt, builder->elements_offset);
else
memcpy(ret, &builder->cnt, sizeof(builder->cnt));
elem = ret + builder->elements_offset;
str = elem + builder->element_size * builder->cnt;
builder->last_numbers =
psync_list_element(builder->number_list.next, psync_list_num_list, list);
builder->popoff = 0;
psync_list_for_each_element(el, &builder->element_list,
psync_list_element_list, list) {
for (i = 0; i < el->used; i++) {
memcpy(elem, el->elements + (i * builder->element_size),
builder->element_size);
scnt = psync_list_bulder_pop_num(builder);
for (j = 0; j < scnt; j++) {
offset = psync_list_bulder_pop_num(builder);
length = psync_list_bulder_pop_num(builder);
pstr = (char **)(elem + offset);
memcpy(str, *pstr, length);
*pstr = str;
str += length;
}
elem += builder->element_size;
}
}
psync_list_for_each_element_call(&builder->element_list, psync_list_element_list, list, free);
psync_list_for_each_element_call(&builder->string_list, psync_list_string_list, list, free);
psync_list_for_each_element_call(&builder->number_list, psync_list_num_list, list, free);
free(builder);
return ret;
}

48
pclsync/plist.h
View File

@ -43,6 +43,47 @@ typedef struct _psync_list {
struct _psync_list *prev;
} psync_list;
typedef struct {
psync_list list;
unsigned long used;
char elements[];
} psync_list_element_list;
typedef struct {
psync_list list;
char *next;
char *end;
} psync_list_string_list;
typedef struct {
psync_list list;
unsigned long used;
uint32_t numbers[1000];
} psync_list_num_list;
struct psync_list_builder_t_ {
size_t element_size;
size_t elements_offset;
size_t elements_per_list;
size_t stringalloc;
uint64_t cnt;
psync_list element_list;
psync_list_element_list *last_elements;
psync_list string_list;
psync_list_string_list *last_strings;
psync_list number_list;
psync_list_num_list *last_numbers;
unsigned long popoff;
char *current_element;
uint32_t *cstrcnt;
};
struct psync_list_builder_t_;
typedef struct psync_list_builder_t_ psync_list_builder_t;
typedef int (*psync_list_compare)(const psync_list *, const psync_list *);
#define psync_list_init(l) \
@ -106,9 +147,8 @@ static inline psync_list *psync_list_remove_head(psync_list *l) {
psync_list_element(psync_list_remove_head(l), t, n)
void psync_list_sort(psync_list *l, psync_list_compare cmp);
void psync_list_extract_repeating(psync_list *l1, psync_list *l2,
psync_list *extracted1,
psync_list *extracted2,
psync_list_compare cmp);
void psync_list_extract_repeating(psync_list *l1, psync_list *l2, psync_list *extracted1, psync_list *extracted2, psync_list_compare cmp);
uint32_t *psync_list_bulder_push_num(psync_list_builder_t *builder);
uint32_t psync_list_bulder_pop_num(psync_list_builder_t *builder);
#endif

1
pclsync/psock.c
View File

@ -39,6 +39,7 @@
#include "psettings.h"
#include "psock.h"
#include "ptimer.h"
#include "ptask.h"
#define PROXY_NONE 0
#define PROXY_CONNECT 1

158
pclsync/ptask.c
View File

@ -58,6 +58,15 @@
#define get_len(t) (sizeof(t) - offsetof(t, request))
#define PSYNC_TASK_STATUS_RUNNING 0
#define PSYNC_TASK_STATUS_READY 1
#define PSYNC_TASK_STATUS_DONE 2
#define PSYNC_TASK_STATUS_RETURNED 3
// #define PSYNC_WAIT_ANYBODY -1 // unused, but may be important later
#define PSYNC_WAIT_NOBODY -2
#define PSYNC_WAIT_FREED -3
#define TASK_TYPE_EXIT 0
#define TASK_TYPE_FILE_DWL 1
#define TASK_TYPE_FILE_DWL_NM 2
@ -159,6 +168,22 @@ typedef struct {
int fd;
} download_context_t;
struct psync_task_t_ {
psync_task_callback_t callback;
void *param;
pthread_cond_t cond;
int id;
int status;
};
struct psync_task_manager_t_ {
pthread_mutex_t mutex;
int taskcnt;
int refcnt;
int waitfor;
struct psync_task_t_ tasks[];
};
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static int running = 0;
static int sockd = INVALID_SOCKET;
@ -825,6 +850,35 @@ static int task_send_async(const void *task, size_t len) {
return ret;
}
static void psync_task_destroy(psync_task_manager_t tm) {
int i;
for (i = 0; i < tm->taskcnt; i++)
pthread_cond_destroy(&tm->tasks[i].cond);
pthread_mutex_destroy(&tm->mutex);
free(tm);
}
static void psync_task_dec_refcnt(psync_task_manager_t tm) {
int refcnt;
pthread_mutex_lock(&tm->mutex);
refcnt = --tm->refcnt;
pthread_mutex_unlock(&tm->mutex);
if (!refcnt)
psync_task_destroy(tm);
}
static psync_task_manager_t psync_get_manager_of_task(struct psync_task_t_ *t) {
return (psync_task_manager_t)(((char *)(t - t->id)) -
offsetof(struct psync_task_manager_t_, tasks));
}
static void psync_task_entry(void *ptr) {
struct psync_task_t_ *t;
t = (struct psync_task_t_ *)ptr;
t->callback(ptr, t->param);
psync_task_dec_refcnt(psync_get_manager_of_task(t));
}
void ptask_ldir_mk(psync_syncid_t syncid,
psync_folderid_t folderid,
psync_folderid_t localfolderid) {
@ -1081,4 +1135,106 @@ void ptask_cfldr_save_filekey(void *ptr) {
psync_sql_run_free(res);
free(t->key);
free(t);
}
}
// from plibs.c
psync_task_manager_t psync_task_run_tasks(psync_task_callback_t const *callbacks, void *const *params, int cnt) {
psync_task_manager_t ret;
struct psync_task_t_ *t;
int i;
ret = (psync_task_manager_t)malloc(
offsetof(struct psync_task_manager_t_, tasks) +
sizeof(struct psync_task_t_) * cnt);
pthread_mutex_init(&ret->mutex, NULL);
ret->taskcnt = cnt;
ret->refcnt = cnt + 1;
ret->waitfor = PSYNC_WAIT_NOBODY;
for (i = 0; i < cnt; i++) {
t = &ret->tasks[i];
t->callback = callbacks[i];
t->param = params[i];
pthread_cond_init(&t->cond, NULL);
t->id = i;
t->status = PSYNC_TASK_STATUS_RUNNING;
prun_thread1("task", psync_task_entry, t);
}
return ret;
}
void *psync_task_papi_result(psync_task_manager_t tm, int id) {
void *ret;
pthread_mutex_lock(&tm->mutex);
if (tm->tasks[id].status == PSYNC_TASK_STATUS_RUNNING) {
do {
tm->waitfor = id;
pthread_cond_wait(&tm->tasks[id].cond, &tm->mutex);
tm->waitfor = PSYNC_WAIT_NOBODY;
} while (tm->tasks[id].status == PSYNC_TASK_STATUS_RUNNING);
ret = tm->tasks[id].param;
tm->tasks[id].status = PSYNC_TASK_STATUS_DONE;
} else if (tm->tasks[id].status == PSYNC_TASK_STATUS_READY) {
ret = tm->tasks[id].param;
tm->tasks[id].status = PSYNC_TASK_STATUS_DONE;
pthread_cond_signal(&tm->tasks[id].cond);
} else {
pdbg_logf(D_BUG, "invalid status %d of task id %d", (int)tm->tasks[id].status,
id);
ret = NULL;
}
pthread_mutex_unlock(&tm->mutex);
return ret;
}
void psync_task_free(psync_task_manager_t tm) {
if (tm->refcnt == 1) {
psync_task_destroy(tm);
}
else {
int refcnt, i;
pthread_mutex_lock(&tm->mutex);
tm->waitfor = PSYNC_WAIT_FREED;
for (i = 0; i < tm->taskcnt; i++)
if (tm->tasks[i].status == PSYNC_TASK_STATUS_READY) {
tm->tasks[i].status = PSYNC_TASK_STATUS_RETURNED;
pthread_cond_signal(&tm->tasks[i].cond);
}
refcnt = --tm->refcnt;
pthread_mutex_unlock(&tm->mutex);
if (!refcnt) {
psync_task_destroy(tm);
}
}
}
int psync_task_complete(void *h, void *data) {
psync_task_manager_t tm;
struct psync_task_t_ *t;
int ret;
t = (struct psync_task_t_ *)h;
tm = psync_get_manager_of_task(t);
pthread_mutex_lock(&tm->mutex);
if (tm->waitfor == t->id) {
t->param = data;
t->status = PSYNC_TASK_STATUS_READY;
pthread_cond_signal(&t->cond);
ret = 0;
} else if (tm->waitfor == PSYNC_WAIT_NOBODY || tm->waitfor >= 0) {
t->param = data;
t->status = PSYNC_TASK_STATUS_READY;
do {
pthread_cond_wait(&t->cond, &tm->mutex);
} while (t->status == PSYNC_TASK_STATUS_READY);
if (t->status == PSYNC_TASK_STATUS_RETURNED)
ret = -1;
else
ret = 0;
} else if (tm->waitfor == PSYNC_WAIT_FREED)
ret = -1;
else {
pdbg_logf(D_BUG, "invalid waitfor value %d", tm->waitfor);
ret = -1;
}
pthread_mutex_unlock(&tm->mutex);
return ret;
}

10
pclsync/ptask.h
View File

@ -127,7 +127,12 @@ typedef struct {
psync_folderid_t id;
} insert_folder_key_task;
struct psync_task_manager_t_;
typedef struct psync_task_manager_t_ *psync_task_manager_t;
typedef void (*psync_async_callback_t)(void *, psync_async_result_t *);
typedef void (*psync_task_callback_t)(void *, void *);
/* Important! The interface typically expect all passed pointers to be alive
@ -166,4 +171,9 @@ void ptask_upload_q(psync_syncid_t syncid, psync_fileid_t localfileid, const cha
void ptask_cfldr_save_fldrkey(void *ptr);
void ptask_cfldr_save_filekey(void *ptr);
psync_task_manager_t psync_task_run_tasks(psync_task_callback_t const *callbacks, void *const *params, int cnt);
void *psync_task_papi_result(psync_task_manager_t tm, int id);
void psync_task_free(psync_task_manager_t tm);
int psync_task_complete(void *h, void *data);
#endif