/* tp.c * Copyright (c) 2012, Peter Ohler * 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 Peter Ohler 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 HOLDER 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. */ /*** This is just a prototype. Turns out the waiting for processing is expensive so the approach is slower that just *** parsing with callbacks. The parsing is only 10% or less of the processing so not much gain would be expected even *** in the best case. ***/ #include #if !IS_WINDOWS #include // for getrlimit() on linux #endif #include #include #include #include #include #include "oj.h" typedef enum { UNDEF = 0, HASH_START, HASH_END, ARRAY_START, ARRAY_END, ADD_STR, ADD_BIG, ADD_FLOAT, ADD_FIXNUM, ADD_TRUE, ADD_FALSE, ADD_NIL, DONE } OpType; typedef struct _Op { OpType op; const char *key; union { VALUE rval; double d; uint64_t i; const char *s; }; } *Op; typedef struct _CX { char *json; VALUE *cur; VALUE *end; struct _Op ops[4096]; Op ops_end; Op iop; VALUE stack[1024]; } *CX; typedef struct _Wait { pthread_mutex_t mutex; pthread_cond_t cond; volatile int waiting; } *Wait; static void hash_start(void *context, const char *key); static void hash_end(void *context, const char *key); static void array_start(void *context, const char *key); static void array_end(void *context, const char *key); static void add_str(void *context, const char *str, const char *key); static void add_big(void *context, const char *str, const char *key); static void add_float(void *context, double num, const char *key); static void add_fixnum(void *context, int64_t num, const char *key); static void add_true(void *context, const char *key); static void add_false(void *context, const char *key); static void add_nil(void *context, const char *key); static void done(void *context); static struct _SajCB tcb = { hash_start, hash_end, array_start, array_end, add_str, add_big, add_float, add_fixnum, add_true, add_false, add_nil, done }; static inline void wakeup(Wait w) { if (w->waiting) { pthread_mutex_lock(&w->mutex); pthread_cond_signal(&w->cond); pthread_mutex_unlock(&w->mutex); } } static int wait_init(Wait w) { int err; w->waiting = 0; if (0 != (err = pthread_mutex_init(&w->mutex, 0)) || 0 != (err = pthread_cond_init(&w->cond, 0))) { return err; } return 0; } static void wait_for_it(Wait w, double timeout) { struct timespec done; struct timeval tv; double end; gettimeofday(&tv, 0); end = (double)tv.tv_sec + (double)tv.tv_usec / 1000000.0; end += timeout; done.tv_sec = (time_t)end; done.tv_nsec = (long)((end - done.tv_sec) * 1000000000.0); pthread_mutex_lock(&w->mutex); w->waiting = 1; pthread_cond_timedwait(&w->cond, &w->mutex, &done); w->waiting = 0; pthread_mutex_unlock(&w->mutex); } static int wait_destroy(Wait w) { int err; if (0 != (err = pthread_mutex_destroy(&w->mutex))) { return err; } if (0 != (err = pthread_cond_destroy(&w->cond))) { return err; } return 0; } static void cx_add(CX cx, VALUE obj, const char *key) { if (0 == cx->cur) { cx->cur = cx->stack; *cx->cur = obj; } else { if (0 != key) { VALUE ks = rb_str_new2(key); #if HAS_ENCODING_SUPPORT rb_enc_associate(ks, oj_utf8_encoding); #endif rb_hash_aset(*cx->cur, ks, obj); } else { rb_ary_push(*cx->cur, obj); } } } static void cx_push(CX cx, VALUE obj, const char *key) { if (0 == cx->cur) { cx->cur = cx->stack; } else { if (cx->end <= cx->cur) { rb_raise(oj_parse_error_class, "too deeply nested"); } cx_add(cx, obj, key); cx->cur++; } *cx->cur = obj; } static pthread_t parse_thread; static CX job = 0; static struct _Wait parse_wait; static struct _Wait proc_wait; static void* parse_loop(void *data) { while (1) { if (0 == job) { wait_for_it(&parse_wait, 0.1); } if (0 != job) { CX cx = job; job = 0; oj_saj_parse(cx->json, &tcb, cx); } } return 0; } void oj_tp_init() { wait_init(&parse_wait); wait_init(&proc_wait); pthread_create(&parse_thread, 0, parse_loop, (void*)job); } VALUE oj_t_parse(char *json) { // TBD change to allocated for each thread or each call, make smaller and block parser when full struct _CX cx; Op op; cx.json = json; cx.cur = 0; cx.end = cx.stack + (sizeof(cx.stack) / sizeof(*cx.stack)); *cx.stack = Qnil; cx.ops_end = cx.ops + (sizeof(cx.ops) / sizeof(*cx.ops)); cx.iop = cx.ops; op = cx.ops; op->op = UNDEF; #if 1 job = &cx; wakeup(&parse_wait); #else oj_saj_parse(cx.json, &tcb, &cx); #endif for (; op < cx.ops_end; op++) { while (UNDEF == op->op) { wait_for_it(&proc_wait, 0.0001); } switch (op->op) { case HASH_START: cx_push(&cx, rb_hash_new(), op->key); break; case HASH_END: cx.cur--; break; case ARRAY_START: cx_push(&cx, rb_ary_new(), op->key); break; case ARRAY_END: cx.cur--; break; case ADD_STR: { VALUE s; s = rb_str_new2(op->s); #if HAS_ENCODING_SUPPORT rb_enc_associate(s, oj_utf8_encoding); #endif cx_add(&cx, s, op->key); } break; case ADD_BIG: cx_add(&cx, rb_funcall(oj_bigdecimal_class, oj_new_id, 1, rb_str_new2(op->s)), op->key); break; case ADD_FLOAT: cx_add(&cx, rb_float_new(op->d), op->key); break; case ADD_FIXNUM: cx_add(&cx, LONG2NUM(op->i), op->key); break; case ADD_TRUE: cx_add(&cx, Qtrue, op->key); break; case ADD_FALSE: cx_add(&cx, Qfalse, op->key); break; case ADD_NIL: cx_add(&cx, Qnil, op->key); break; case DONE: return *cx.stack; default: // TBD raise printf("*** unknown op type %d\n", op->op); return Qnil; } } return *cx.stack; } inline static void append_op(CX cx, OpType op, const char *key) { (cx->iop + 1)->op = UNDEF; cx->iop->key = key; cx->iop->op = op; wakeup(&proc_wait); cx->iop++; } static void hash_start(void *context, const char *key) { append_op((CX)context, HASH_START, key); } static void hash_end(void *context, const char *key) { append_op((CX)context, HASH_END, key); } static void array_start(void *context, const char *key) { append_op((CX)context, ARRAY_START, key); } static void array_end(void *context, const char *key) { append_op((CX)context, ARRAY_END, key); } static void add_str(void *context, const char *str, const char *key) { CX cx = (CX)context; cx->iop->s = str; append_op(cx, ADD_STR, key); } static void add_big(void *context, const char *str, const char *key) { CX cx = (CX)context; cx->iop->s = str; append_op(cx, ADD_BIG, key); } static void add_float(void *context, double num, const char *key) { CX cx = (CX)context; cx->iop->d = num; append_op(cx, ADD_FLOAT, key); } static void add_fixnum(void *context, int64_t num, const char *key) { CX cx = (CX)context; cx->iop->i = num; append_op(cx, ADD_FIXNUM, key); } static void add_true(void *context, const char *key) { append_op((CX)context, ADD_TRUE, key); } static void add_false(void *context, const char *key) { append_op((CX)context, ADD_FALSE, key); } static void add_nil(void *context, const char *key) { append_op((CX)context, ADD_NIL, key); } static void done(void *context) { append_op((CX)context, DONE, 0); }