1/**
2 * @copyright 2013 Couchbase, Inc.
3 *
4 * @author Filipe Manana  <filipe@couchbase.com>
5 *
6 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
7 * use this file except in compliance with the License. You may obtain a copy of
8 * the License at
9 *
10 *  http://www.apache.org/licenses/LICENSE-2.0
11 *
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15 * License for the specific language governing permissions and limitations under
16 * the License.
17 **/
18
19#include "mapreduce.h"
20#include "mapreduce_internal.h"
21#include <iostream>
22#include <cstring>
23#include <platform/cb_malloc.h>
24#include <stdlib.h>
25// This is libv8_libplatform library which handles garbage collection for v8
26#include <libplatform/libplatform.h>
27
28using namespace v8;
29
30typedef struct {
31    Persistent<Object>    jsonObject;
32    Persistent<Function>  jsonParseFun;
33    Persistent<Function>  stringifyFun;
34    mapreduce_ctx_t       *ctx;
35} isolate_data_t;
36
37
38static const char *SUM_FUNCTION_STRING =
39    "(function(values) {"
40    "    var sum = 0;"
41    "    for (var i = 0; i < values.length; ++i) {"
42    "        sum += values[i];"
43    "    }"
44    "    return sum;"
45    "})";
46
47static const char *DATE_FUNCTION_STRING =
48    // I wish it was on the prototype, but that will require bigger
49    // C changes as adding to the date prototype should be done on
50    // process launch. The code you see here may be faster, but it
51    // is less JavaScripty.
52    // "Date.prototype.toArray = (function() {"
53    "(function(date) {"
54    "    date = date.getUTCDate ? date : new Date(date);"
55    "    return isFinite(date.valueOf()) ?"
56    "      [date.getUTCFullYear(),"
57    "      (date.getUTCMonth() + 1),"
58    "       date.getUTCDate(),"
59    "       date.getUTCHours(),"
60    "       date.getUTCMinutes(),"
61    "       date.getUTCSeconds()] : null;"
62    "})";
63
64static const char *BASE64_FUNCTION_STRING =
65    "(function(b64) {"
66    "    var i, j, l, tmp, scratch, arr = [];"
67    "    var lookup = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/';"
68    "    if (typeof b64 !== 'string') {"
69    "        throw 'Input is not a string';"
70    "    }"
71    "    if (b64.length % 4 > 0) {"
72    "        throw 'Invalid base64 source.';"
73    "    }"
74    "    scratch = b64.indexOf('=');"
75    "    scratch = scratch > 0 ? b64.length - scratch : 0;"
76    "    l = scratch > 0 ? b64.length - 4 : b64.length;"
77    "    for (i = 0, j = 0; i < l; i += 4, j += 3) {"
78    "        tmp = (lookup.indexOf(b64[i]) << 18) | (lookup.indexOf(b64[i + 1]) << 12);"
79    "        tmp |= (lookup.indexOf(b64[i + 2]) << 6) | lookup.indexOf(b64[i + 3]);"
80    "        arr.push((tmp & 0xFF0000) >> 16);"
81    "        arr.push((tmp & 0xFF00) >> 8);"
82    "        arr.push(tmp & 0xFF);"
83    "    }"
84    "    if (scratch === 2) {"
85    "        tmp = (lookup.indexOf(b64[i]) << 2) | (lookup.indexOf(b64[i + 1]) >> 4);"
86    "        arr.push(tmp & 0xFF);"
87    "    } else if (scratch === 1) {"
88    "        tmp = (lookup.indexOf(b64[i]) << 10) | (lookup.indexOf(b64[i + 1]) << 4);"
89    "        tmp |= (lookup.indexOf(b64[i + 2]) >> 2);"
90    "        arr.push((tmp >> 8) & 0xFF);"
91    "        arr.push(tmp & 0xFF);"
92    "    }"
93    "    return arr;"
94    "})";
95
96
97
98static Local<Context> createJsContext();
99static void emit(const FunctionCallbackInfo<Value> &args);
100
101static void doInitContext(mapreduce_ctx_t *ctx);
102static Handle<Function> compileFunction(const std::string &function);
103static std::string exceptionString(const TryCatch &tryCatch);
104static void loadFunctions(mapreduce_ctx_t *ctx,
105                          const std::list<std::string> &function_sources);
106static inline isolate_data_t *getIsolateData();
107static inline mapreduce_json_t jsonStringify(const Handle<Value> &obj);
108static inline Handle<Value> jsonParse(const mapreduce_json_t &thing);
109static inline void taskStarted(mapreduce_ctx_t *ctx);
110static inline void taskFinished(mapreduce_ctx_t *ctx);
111static void freeKvListEntries(kv_list_int_t &kvs);
112static void freeJsonListEntries(json_results_list_t &list);
113static inline Handle<Array> jsonListToJsArray(const mapreduce_json_list_t &list);
114
115static Platform *v8platform;
116void initV8()
117{
118    V8::InitializeICU();
119    v8platform = platform::CreateDefaultPlatform();
120    V8::InitializePlatform(v8platform);
121    V8::Initialize();
122}
123
124void deinitV8()
125{
126    V8::Dispose();
127    V8::ShutdownPlatform();
128    delete v8platform;
129}
130
131void initContext(mapreduce_ctx_t *ctx,
132                 const std::list<std::string> &function_sources)
133{
134    doInitContext(ctx);
135
136    try {
137        Locker locker(ctx->isolate);
138        Isolate::Scope isolate_scope(ctx->isolate);
139        HandleScope handle_scope(ctx->isolate);
140        Local<Context> context =
141            Local<Context>::New(ctx->isolate, ctx->jsContext);
142        Context::Scope context_scope(context);
143
144        loadFunctions(ctx, function_sources);
145    } catch (...) {
146        destroyContext(ctx);
147        throw;
148    }
149}
150
151
152void destroyContext(mapreduce_ctx_t *ctx)
153{
154    {
155        Locker locker(ctx->isolate);
156        Isolate::Scope isolate_scope(ctx->isolate);
157        HandleScope handle_scope(ctx->isolate);
158        Local<Context> context =
159            Local<Context>::New(ctx->isolate, ctx->jsContext);
160        Context::Scope context_scope(context);
161
162        for (unsigned int i = 0; i < ctx->functions->size(); ++i) {
163            (*ctx->functions)[i]->Reset();
164            delete (*ctx->functions)[i];
165
166        }
167        delete ctx->functions;
168
169        isolate_data_t *isoData = getIsolateData();
170        isoData->jsonObject.Reset();
171        isoData->jsonParseFun.Reset();
172        isoData->stringifyFun.Reset();
173        delete isoData;
174
175        ctx->jsContext.Reset();
176    }
177
178    ctx->isolate->Dispose();
179
180}
181
182static Local<String> createUtf8String(Isolate *isolate, const char *str)
183{
184    return String::NewFromUtf8(isolate, str,
185        NewStringType::kNormal).ToLocalChecked();
186}
187
188static Local<String> createUtf8String(Isolate *isolate, const char *str,
189                                      size_t len)
190{
191    return String::NewFromUtf8(isolate, str,
192        NewStringType::kNormal, len).ToLocalChecked();
193}
194
195static void doInitContext(mapreduce_ctx_t *ctx)
196{
197    Isolate::CreateParams createParams;
198    std::unique_ptr<ArrayBuffer::Allocator> allocator(
199      ArrayBuffer::Allocator::NewDefaultAllocator());
200    createParams.array_buffer_allocator =
201      allocator.get();
202    ctx->isolate = Isolate::New(createParams);
203    Locker locker(ctx->isolate);
204    Isolate::Scope isolate_scope(ctx->isolate);
205
206    HandleScope handle_scope(ctx->isolate);
207    ctx->jsContext.Reset(ctx->isolate, createJsContext());
208    Local<Context> context = Local<Context>::New(ctx->isolate, ctx->jsContext);
209    Context::Scope context_scope(context);
210    Local<String> jsonString = createUtf8String(ctx->isolate, "JSON");
211    Handle<Object> jsonObject =
212        Local<Object>::Cast(context->Global()->Get(jsonString));
213
214    Local<String> parseString = createUtf8String(ctx->isolate, "parse");
215    Handle<Function> parseFun =
216        Local<Function>::Cast(jsonObject->Get(parseString));
217    Local<String> stringifyString = createUtf8String(ctx->isolate, "stringify");
218    Handle<Function> stringifyFun =
219        Local<Function>::Cast(jsonObject->Get(stringifyString));
220
221    isolate_data_t *isoData = new isolate_data_t();
222    isoData->jsonObject.Reset(ctx->isolate, jsonObject);
223    isoData->jsonParseFun.Reset(ctx->isolate, parseFun);
224    isoData->stringifyFun.Reset(ctx->isolate, stringifyFun);
225    isoData->ctx = ctx;
226
227    ctx->isolate->SetData(0, (void *)isoData);
228    ctx->taskStartTime = -1;
229}
230
231
232static Local<Context> createJsContext()
233{
234    Isolate *isolate = Isolate::GetCurrent();
235    EscapableHandleScope handle_scope(isolate);
236
237    Handle<ObjectTemplate> global = ObjectTemplate::New();
238    global->Set(createUtf8String(isolate, "emit"),
239            FunctionTemplate::New(isolate, emit));
240
241    Handle<Context> context = Context::New(isolate, NULL, global);
242    Context::Scope context_scope(context);
243
244    Handle<Function> sumFun = compileFunction(SUM_FUNCTION_STRING);
245    context->Global()->Set(createUtf8String(isolate, "sum"), sumFun);
246
247    Handle<Function> decodeBase64Fun =
248        compileFunction(BASE64_FUNCTION_STRING);
249    context->Global()->Set(createUtf8String(isolate, "decodeBase64"),
250        decodeBase64Fun);
251
252    Handle<Function> dateToArrayFun =
253        compileFunction(DATE_FUNCTION_STRING);
254    context->Global()->Set(createUtf8String(isolate, "dateToArray"),
255                           dateToArrayFun);
256
257    // Use EscapableHandleScope and return using .Escape
258    // This will ensure that return values are not garbage collected
259    // as soon as the function returns.
260    return handle_scope.Escape(context);
261}
262
263
264void mapDoc(mapreduce_ctx_t *ctx,
265            const mapreduce_json_t &doc,
266            const mapreduce_json_t &meta,
267            mapreduce_map_result_list_t *results)
268{
269    Locker locker(ctx->isolate);
270    Isolate::Scope isolate_scope(ctx->isolate);
271    HandleScope handle_scope(ctx->isolate);
272    Local<Context> context = Local<Context>::New(ctx->isolate, ctx->jsContext);
273    Context::Scope context_scope(context);
274    Handle<Value> docObject = jsonParse(doc);
275    Handle<Value> metaObject = jsonParse(meta);
276
277
278    if (!metaObject->IsObject()) {
279        throw MapReduceError(MAPREDUCE_INVALID_ARG,
280                "metadata is not a JSON object");
281    }
282
283    Handle<Value> funArgs[] = { docObject, metaObject };
284
285    taskStarted(ctx);
286    kv_list_int_t kvs;
287    ctx->kvs = &kvs;
288
289    for (unsigned int i = 0; i < ctx->functions->size(); ++i) {
290        mapreduce_map_result_t mapResult;
291        Local<Function> fun =
292            Local<Function>::New(ctx->isolate, *(*ctx->functions)[i]);
293        TryCatch try_catch(ctx->isolate);
294        Handle<Value> result = fun->Call(context->Global(), 2, funArgs);
295
296        if (!result.IsEmpty()) {
297            mapResult.error = MAPREDUCE_SUCCESS;
298            mapResult.result.kvs.length = kvs.size();
299            size_t sz = sizeof(mapreduce_kv_t) * mapResult.result.kvs.length;
300            mapResult.result.kvs.kvs = (mapreduce_kv_t *) cb_malloc(sz);
301            if (mapResult.result.kvs.kvs == NULL) {
302                freeKvListEntries(kvs);
303                throw std::bad_alloc();
304            }
305            kv_list_int_t::iterator it = kvs.begin();
306            for (int j = 0; it != kvs.end(); ++it, ++j) {
307                mapResult.result.kvs.kvs[j] = *it;
308            }
309        } else {
310            freeKvListEntries(kvs);
311
312            if (!try_catch.CanContinue()) {
313                throw MapReduceError(MAPREDUCE_TIMEOUT, "timeout");
314            }
315
316            mapResult.error = MAPREDUCE_RUNTIME_ERROR;
317            std::string exceptString = exceptionString(try_catch);
318            size_t len = exceptString.length();
319
320            mapResult.result.error_msg = (char *) cb_malloc(len + 1);
321            if (mapResult.result.error_msg == NULL) {
322                throw std::bad_alloc();
323            }
324            memcpy(mapResult.result.error_msg, exceptString.data(), len);
325            mapResult.result.error_msg[len] = '\0';
326        }
327
328        results->list[i] = mapResult;
329        results->length += 1;
330        kvs.clear();
331    }
332
333    taskFinished(ctx);
334}
335
336
337json_results_list_t runReduce(mapreduce_ctx_t *ctx,
338                              const mapreduce_json_list_t &keys,
339                              const mapreduce_json_list_t &values)
340{
341    Locker locker(ctx->isolate);
342    Isolate::Scope isolateScope(ctx->isolate);
343    HandleScope handle_scope(ctx->isolate);
344    Local<Context> context = Local<Context>::New(ctx->isolate, ctx->jsContext);
345    Context::Scope context_scope(context);
346    Handle<Array> keysArray = jsonListToJsArray(keys);
347    Handle<Array> valuesArray = jsonListToJsArray(values);
348    json_results_list_t results;
349
350    Handle<Value> args[] =
351        { keysArray, valuesArray, Boolean::New(ctx->isolate, false) };
352
353    taskStarted(ctx);
354
355    for (unsigned int i = 0; i < ctx->functions->size(); ++i) {
356        Local<Function> fun =
357            Local<Function>::New(ctx->isolate, *(*ctx->functions)[i]);
358        TryCatch try_catch(ctx->isolate);
359        Handle<Value> result = fun->Call(context->Global(), 3, args);
360
361        if (result.IsEmpty()) {
362            freeJsonListEntries(results);
363
364            if (!try_catch.CanContinue()) {
365                throw MapReduceError(MAPREDUCE_TIMEOUT, "timeout");
366            }
367
368            throw MapReduceError(MAPREDUCE_RUNTIME_ERROR,
369                    exceptionString(try_catch));
370        }
371
372        try {
373            mapreduce_json_t jsonResult = jsonStringify(result);
374            results.push_back(jsonResult);
375        } catch(...) {
376            freeJsonListEntries(results);
377            throw;
378        }
379    }
380
381    taskFinished(ctx);
382
383    return results;
384}
385
386
387mapreduce_json_t runReduce(mapreduce_ctx_t *ctx,
388                           int reduceFunNum,
389                           const mapreduce_json_list_t &keys,
390                           const mapreduce_json_list_t &values)
391{
392    Locker locker(ctx->isolate);
393    Isolate::Scope isolateScope(ctx->isolate);
394    HandleScope handle_scope(ctx->isolate);
395    Local<Context> context = Local<Context>::New(ctx->isolate, ctx->jsContext);
396    Context::Scope context_scope(context);
397
398    reduceFunNum -= 1;
399    if (reduceFunNum < 0 ||
400        static_cast<unsigned int>(reduceFunNum) >= ctx->functions->size()) {
401        throw MapReduceError(MAPREDUCE_INVALID_ARG,
402                "invalid reduce function number");
403    }
404
405    Local<Function> fun =
406        Local<Function>::New(ctx->isolate, *(*ctx->functions)[reduceFunNum]);
407    Handle<Array> keysArray = jsonListToJsArray(keys);
408    Handle<Array> valuesArray = jsonListToJsArray(values);
409    Handle<Value> args[] =
410        { keysArray, valuesArray, Boolean::New(ctx->isolate, false) };
411
412    taskStarted(ctx);
413
414    TryCatch try_catch(ctx->isolate);
415    Handle<Value> result = fun->Call(context->Global(), 3, args);
416
417    taskFinished(ctx);
418
419    if (result.IsEmpty()) {
420        if (!try_catch.CanContinue()) {
421            throw MapReduceError(MAPREDUCE_TIMEOUT, "timeout");
422        }
423
424        throw MapReduceError(MAPREDUCE_RUNTIME_ERROR,
425                exceptionString(try_catch));
426    }
427
428    return jsonStringify(result);
429}
430
431
432mapreduce_json_t runRereduce(mapreduce_ctx_t *ctx,
433                             int reduceFunNum,
434                             const mapreduce_json_list_t &reductions)
435{
436    Locker locker(ctx->isolate);
437    Isolate::Scope isolateScope(ctx->isolate);
438    HandleScope handle_scope(ctx->isolate);
439    Local<Context> context = Local<Context>::New(ctx->isolate, ctx->jsContext);
440    Context::Scope context_scope(context);
441
442    reduceFunNum -= 1;
443    if (reduceFunNum < 0 ||
444        static_cast<unsigned int>(reduceFunNum) >= ctx->functions->size()) {
445        throw MapReduceError(MAPREDUCE_INVALID_ARG,
446                "invalid reduce function number");
447    }
448
449    Local<Function> fun =
450        Local<Function>::New(ctx->isolate, *(*ctx->functions)[reduceFunNum]);
451    Handle<Array> valuesArray = jsonListToJsArray(reductions);
452    Handle<Value> args[] =
453        { Null(ctx->isolate), valuesArray, Boolean::New(ctx->isolate, true) };
454
455    taskStarted(ctx);
456
457    TryCatch try_catch(ctx->isolate);
458    Handle<Value> result = fun->Call(context->Global(), 3, args);
459
460    taskFinished(ctx);
461
462    if (result.IsEmpty()) {
463        if (!try_catch.CanContinue()) {
464            throw MapReduceError(MAPREDUCE_TIMEOUT, "timeout");
465        }
466
467        throw MapReduceError(MAPREDUCE_RUNTIME_ERROR,
468                exceptionString(try_catch));
469    }
470
471    return jsonStringify(result);
472}
473
474
475void terminateTask(mapreduce_ctx_t *ctx)
476{
477    V8::TerminateExecution(ctx->isolate);
478    ctx->taskStartTime = -1;
479}
480
481
482static void freeKvListEntries(kv_list_int_t &kvs)
483{
484    kv_list_int_t::iterator it = kvs.begin();
485
486    for ( ; it != kvs.end(); ++it) {
487        mapreduce_kv_t kv = *it;
488        cb_free(kv.key.json);
489        cb_free(kv.value.json);
490    }
491    kvs.clear();
492}
493
494
495static void freeJsonListEntries(json_results_list_t &list)
496{
497    json_results_list_t::iterator it = list.begin();
498
499    for ( ; it != list.end(); ++it) {
500        cb_free((*it).json);
501    }
502    list.clear();
503}
504
505
506static Handle<Function> compileFunction(const std::string &funSource)
507{
508    Isolate *isolate = Isolate::GetCurrent();
509    Local<Context> context(isolate->GetCurrentContext());
510    EscapableHandleScope handle_scope(isolate);
511    TryCatch try_catch(isolate);
512    Local<String> source = createUtf8String(isolate, funSource.data(),
513        funSource.length());
514    Local<Script> script;
515    if (!Script::Compile(context, source).ToLocal(&script)) {
516        throw MapReduceError(MAPREDUCE_SYNTAX_ERROR,
517                exceptionString(try_catch));
518    }
519
520    if (script.IsEmpty()) {
521        throw MapReduceError(MAPREDUCE_SYNTAX_ERROR,
522                exceptionString(try_catch));
523    }
524
525    Handle<Value> result = script->Run();
526
527    if (result.IsEmpty()) {
528        throw MapReduceError(MAPREDUCE_SYNTAX_ERROR,
529                exceptionString(try_catch));
530    }
531
532    if (!result->IsFunction()) {
533        throw MapReduceError(MAPREDUCE_SYNTAX_ERROR,
534                std::string("Invalid function: ") + funSource.c_str());
535    }
536
537    // Use EscapableHandleScope and return using .Escape
538    // This will ensure that return values are not garbage collected
539    // as soon as the function returns.
540    return handle_scope.Escape(Handle<Function>::Cast(result));
541}
542
543
544static std::string exceptionString(const TryCatch &tryCatch)
545{
546    HandleScope handle_scope(Isolate::GetCurrent());
547    String::Utf8Value exception(tryCatch.Exception());
548    const char *exceptionString = (*exception);
549
550    if (exceptionString) {
551        Handle<Message> message = tryCatch.Message();
552        return std::string(exceptionString) + " (line " +
553            std::to_string(message->GetLineNumber()) + ":" +
554            std::to_string(message->GetStartColumn()) + ")";
555    }
556
557    return std::string("runtime error");
558}
559
560
561static void loadFunctions(mapreduce_ctx_t *ctx,
562                          const std::list<std::string> &function_sources)
563{
564    HandleScope handle_scope(ctx->isolate);
565
566    ctx->functions = new function_vector_t();
567
568    std::list<std::string>::const_iterator it = function_sources.begin();
569
570    for ( ; it != function_sources.end(); ++it) {
571        Handle<Function> fun = compileFunction(*it);
572        Persistent<Function> *perFn = new Persistent<Function>();
573        perFn->Reset(ctx->isolate, fun);
574        ctx->functions->push_back(perFn);
575    }
576}
577
578
579static void emit(const FunctionCallbackInfo<Value> &args)
580{
581    isolate_data_t *isoData = getIsolateData();
582
583    if (isoData->ctx->kvs == NULL) {
584        return;
585    }
586
587    try {
588        mapreduce_kv_t result;
589
590        result.key   = jsonStringify(args[0]);
591        result.value = jsonStringify(args[1]);
592        isoData->ctx->kvs->push_back(result);
593
594        return;
595    } catch(Local<String> &ex) {
596        isoData->ctx->isolate->ThrowException(ex);
597    }
598}
599
600
601static inline isolate_data_t *getIsolateData()
602{
603    Isolate *isolate = Isolate::GetCurrent();
604    return reinterpret_cast<isolate_data_t*>(isolate->GetData(0));
605}
606
607
608static inline mapreduce_json_t jsonStringify(const Handle<Value> &obj)
609{
610    isolate_data_t *isoData = getIsolateData();
611    Handle<Value> args[] = { obj };
612    TryCatch try_catch(isoData->ctx->isolate);
613    Local<Function> stringifyFun =
614        Local<Function>::New(isoData->ctx->isolate, isoData->stringifyFun);
615    Local<Object> jsonObject =
616        Local<Object>::New(isoData->ctx->isolate, isoData->jsonObject);
617    Handle<Value> result = stringifyFun->Call(jsonObject, 1, args);
618
619    if (result.IsEmpty()) {
620        throw try_catch.Exception();
621    }
622
623    mapreduce_json_t jsonResult;
624
625    if (!result->IsUndefined()) {
626        Handle<String> str = Handle<String>::Cast(result);
627        jsonResult.length = str->Utf8Length();
628        jsonResult.json = (char *) cb_malloc(jsonResult.length);
629        if (jsonResult.json == NULL) {
630            throw std::bad_alloc();
631        }
632        str->WriteUtf8(jsonResult.json, jsonResult.length,
633                       NULL, String::NO_NULL_TERMINATION);
634    } else {
635        jsonResult.length = sizeof("null") - 1;
636        jsonResult.json = (char *) cb_malloc(jsonResult.length);
637        if (jsonResult.json == NULL) {
638            throw std::bad_alloc();
639        }
640        memcpy(jsonResult.json, "null", jsonResult.length);
641    }
642
643    // Caller responsible for freeing jsonResult.json
644    return jsonResult;
645}
646
647
648static inline Handle<Value> jsonParse(const mapreduce_json_t &thing)
649{
650    isolate_data_t *isoData = getIsolateData();
651    Handle<Value> args[] =
652        { createUtf8String(isoData->ctx->isolate, thing.json,
653            thing.length) };
654    TryCatch try_catch(isoData->ctx->isolate);
655    Local<Function> jsonParseFun =
656        Local<Function>::New(isoData->ctx->isolate, isoData->jsonParseFun);
657    Local<Object> jsonObject =
658        Local<Object>::New(isoData->ctx->isolate, isoData->jsonObject);
659    Handle<Value> result = jsonParseFun->Call(jsonObject, 1, args);
660
661    if (result.IsEmpty()) {
662        throw MapReduceError(MAPREDUCE_RUNTIME_ERROR,
663                exceptionString(try_catch));
664    }
665
666    return result;
667}
668
669
670static inline void taskStarted(mapreduce_ctx_t *ctx)
671{
672    ctx->taskStartTime = time(NULL);
673    ctx->kvs = NULL;
674}
675
676
677static inline void taskFinished(mapreduce_ctx_t *ctx)
678{
679    ctx->exitMutex.lock();
680    ctx->taskStartTime = -1;
681    ctx->exitMutex.unlock();
682}
683
684
685static inline Handle<Array> jsonListToJsArray(const mapreduce_json_list_t &list)
686{
687    Isolate *isolate = Isolate::GetCurrent();
688    Handle<Array> array = Array::New(isolate, list.length);
689
690    for (int i = 0 ; i < list.length; ++i) {
691        Handle<Value> v = jsonParse(list.values[i]);
692        array->Set(Number::New(isolate, i), v);
693    }
694
695    return array;
696}
697