node-sync — псевдо-синхронное программирование на nodejs с использованием fibers

Надавно была опубликована библиотека node-fibers, вносящая в nodejs и v8 поддержку замечательного fiber/coroutine — тоесть, возможность использовать yield.
Параллельно, на nodejs groups прошел целый ряд обсуждений на тему всевозможных вариантов упрощения асинхронного синтаксиса.

Вдохновившись возможностями, которые дают «волокна», я написал библиотеку node-sync, которая делает разработку в асинхронном окружении nodejs значительно удобнее, а код — нагляднее.

Синопсис

// Обычная асинхронная функция, вызывает callback с результатом через 1 сек
function someAsyncFunction(a, b, callback) {
    setTimeout(function(){
        callback(null, a + b);
    }, 1000)
}
// Вызываем эту функцию синхронно, используя Function.prototype.sync(),
// работающий по тому же принципу, что и call()
// на этом моменте текуший поток "зависнет" на секунду, пока функция не вернет значение
var result = someAsyncFunction.sync(null, 2, 3);
console.log(result); // "5" через 1 секунду

Философия

Главная идея в том, что метод Function.prototype.sync() встроен в любую функцию по умолчанию, а так же, его интерфейс соответствует всем давно известному call(). Подключив библиотеку sync, мы можем вызвать любую асинхронную функцию синхронно, без написания дополнительного кода.
Псевдо-синхронное программирование — потому, что фактически, Function.prototype.sync() не блокирует весь процесс, а только текущий поток. Тело самой функции выполняется асинхронно, мы просто дожидаемся результата (используя «yield»). Но при этом, код читается «синхронно».

node-sync решает для меня три важных вопроса:
1. Освобождение от бесконечной индентации с коллбэками (избавляет от «спагетти-кода»)
2. Корректная обработка ошибок
3. Интеграция с существующим кодом/библиотеками без необходимости рефакторинга

Уже больше месяца я использую эту node-sync в своем приложении, переход был незаметным — я просто начал писать новый код на «псевдо-синхронный» манер, старый код остался прежним.

Блокировка

Прелесть «волокон» состоит в том, что при ожидании (yield) блокируется только текущий поток, а не весь процесс. Наглядный пример блокировки всего процесса — fs.readFileSync и другие псевдо-синхронные функции.

Используя «волокна» можно избежать глобальной блокировки и, при этом, прочитать файл синхронно:

var fs = require('fs'),
    Sync = require('sync');

// запускаем новый поток
Sync(function(){ // тело потока -->
    
    // синхронно читаем файл, используя Function.prototype.sync()
    var source = fs.readFile.sync(null, __filename);
    // выводим содержимое текущего файла
    console.log(String(source));
})

Отличие этого кода в том, что пока мы ожидаем ответ от fs.readFile.sync(), приложение спокойно продолжает выполнять другие операции.

Обработка ошибок

Всем, кто хоть раз пробовал написать на nodejs что-то серьезнее, чем «hello world app», наверняка знакома рутина с обработкой ошибок. Если следовать официальному дизайну callback-функций, первым агрументом всегда должна быть возвращена ошибка. При этом, использовать throw в асинхронном окружении чревато падением всего event-loop.

Весьма реальный код на nodejs с корректной обработкой ошибок:

// Функция должна что-то сделать и вернуть результат,
// при этом, корректно обработать ошибку в случае неудачи
function asyncFunction(callback)
{
    var p_client = new Db('test', new Server("127.0.0.1", 27017, {}));
    
    p_client.open(function(err, p_client) {
        if (err) return callback(err); // <-- рутина
        
        p_client.createCollection('test_custom_key', function(err, collection) {
            if (err) return callback(err); // <-- рутина
            
            collection.insert({'a':1}, function(err, docs) {
                if (err) return callback(err); // <-- рутина
                
                collection.find({'_id':new ObjectID("aaaaaaaaaaaa")}, function(err, cursor) {
                    if (err) return callback(err); // <-- рутина
                    
                    cursor.toArray(function(err, items) {
                        if (err) return callback(err); // <-- рутина
                        
                        // результат = items
                        callback(null, items);
                    });
                });
            });
        });
    });
}

Такая-же функция, только использующая sync. Результат ее работы идентичен функции выше, учитывая обработку ошибок. Если какая-то из вызываемых функций вернет ошибку в авто-callback, который передает ей sync(), то эта ошибка прозрачно попадет в результирующий callback, который мы указали потоку вторым аргументом.

function syncFunction(callback)
{
    // запускаем поток
    Sync(function(){
        
        var p_client = new Db('test', new Server("127.0.0.1", 27017, {}));
        p_client.open.sync(p_client);
        
        var collection = p_client.createCollection.sync(p_client, 'test');
        collection.insert.sync(collection, {'a' : 1});
        
        var cursor = collection.find.sync(collection, {'_id':new ObjectID("aaaaaaaaaaaa"))
        var items = cursor.toArray.sync(cursor);
        
        // результат = items
        return items;
        
    }, callback) // <-- результат возвращаем в callback
}

Используя специальный метод Function.prototype.async(), эта функция может быть еще проще (работает аналогично функции выше):

var syncFunction = function()
{
    var p_client = new Db('test', new Server("127.0.0.1", 27017, {}));
    p_client.open.sync(p_client);
    
    var collection = p_client.createCollection.sync(p_client, 'test');
    collection.insert.sync(collection, {'a' : 1});
    
    var cursor = collection.find.sync(collection, {'_id':new ObjectID("aaaaaaaaaaaa"))
    var items = cursor.toArray.sync(cursor);
    
    // результат = items
    return items;
    
}.async() // <-- в этом месте мы превращаем ее в асинхронную функцию

Параллельность

Иногда, нам нужно выполнить несколько функций параллельно, при этом, дождаться всех результатов, и только тогда продолжить. Для этого существует Sync.Parallel:

var Sync = require('sync');

// Какая-то асинхронная функция, возвращает результат через секунду
function someAsyncFunction(a, b, callback) {
    setTimeout(function(){
        callback(null, a + b);
    }, 1000)
}

// Новый поток
Sync(function(){
    
    // Запускаем параллельно функцию с разными аргументами
    // поток будет заблокирован до тех пор, пока не будут возвращены оба результата
    var results = Sync.Parallel(function(callback){
        someAsyncFunction(2, 2, callback());
        someAsyncFunction(5, 5, callback());
    });
    console.log(results); // [ 4, 10 ]
    
    // Ассоциативный вариант
    var results = Sync.Parallel(function(callback){
        someAsyncFunction(2, 2, callback('foo')); // assign the result to 'foo'
        someAsyncFunction(5, 5, callback('bar')); // assign the result to 'bar'
    });
    console.log(results); // { foo: 4, bar: 10 }
})

На днях общался с господином laverdet (создатель node-fibers для v8), и он предложил весьма интересную парадигму «будущего». Я добавил новый метод Function.prototype.future() — его тоже можно использовать для параллельности:

// Новый поток
Sync(function(){
    // Запускаем someAsyncFunction, но не блокируем поток
    var foo = someAsyncFunction.future(null, 2, 2);
    var bar = someAsyncFunction.future(null, 5, 5);
    
    // foo, bar - это билеты в будущее
    console.log(foo); // { [Function: Future] result: [Getter], error: [Getter] }
    
    // А вот теперь, дожидаемся значений от foo и bar
    console.log(foo.result, bar.result); // 4 10 - ровно через секунду (не две)
})

Установка

$ npm install sync
$ node-fibers my_file.js

Имейте ввиду, что для поддержки fibers вам нужно использовать скрипт «node-fibers» вместо «node».

API

var Sync = require('sync');

// Новый поток, fn - функция-тело, результат не возвращается
Sync(fn)

// Новый поток, fn - функция-тело, результат/ошибка возвращается в callback
Sync(fn, callback)

// внутри используется инкрементальный callback()
Sync.Parallel(function(callback){
    callback() // без аргумента (вернется массив)
    callback('foo') // или ассоциативный ключ
}) 

// Вызывает функцию асинхронно и ждет результата/ошибки
// obj - контекст, остальные аргументы попадут по порядку в функцию
Function.prototype.sync(obj, arg1, arg2)

// Вызывает функцию асинхронно и не ждет результата, возвращая управление в контекст
// возвращает объект/функцию Future, у которой есть getter 'result'
// при попытке получения Future.result, поток будет заблокирован до тех пор, пока результат не будет получен
// obj - контекст, остальные аргументы попадут по порядку в функцию
Function.prototype.future(obj, arg1, arg2)

// Делает из любой синхронной функции - асинхронную
// возвращает функцию, которую можно вызвать асинхронно
// obj - контекст
Function.prototype.async(obj)

Резюме

Я и дальше намерен развивать это направление в nodejs, ибо мне оно кажется очень правильным. Буду рад, если кто-то из вас вдохновится этой идеей и внесет свой вклад в ее развитие.

Советую посмотреть довольно подробные примеры использования библиотеки.
Если вы намерены поучавствовать в разработке — форкайте, милости просим, только не забывайте про тесты.
Я так же добавил скрипт в benchmarks. Если у кого-то появятся еще идеи, как можно протестировать скорость работы fibers, будет круто.

Хочу поблагодарить egorF за брейнсторминг, и за то, что вообще заразил меня темой fibers :)

Вам так же могут быть интересны другие библиотеки, основанные на node-fibers.