COMET: серверная часть - паттерны реализации

В этой статье мы рассмотрим распространенные способы создания COMET-сервера и примеры таких серверов на языках Java, Javascript, Python, Erlang.

Основное внимание уделено паттернам серверной поддержки, хотя и примеры реализаций, рассмотренные ниже, вполне расширяемы и жизнеспособны.

Введение. Классические серверные паттерны.

При программировании сайтов, создании динамических страничек и сервисов, как правило, удобен простой цикл обработки запроса:

1. Принять запрос
2. Сгенерировать страничку
3. Выдать страницу
4. Закрыть соединение

При этом для параллельной обработки запросов создается много процессов или потоков, между которыми распределяются приходящие запросы.

Рассмотрим эти два варианта подробнее.

Процессы OS

Самый простой для программирования подход - это обрабатывать запрос в отдельном процессе операционной системы.

В этом случае программа, генерирующая страницы, запускается в множестве экземпляров (процессов), и пришедший запрос отдается тому процессу, который в данный момент свободен.

Можно запускать фиксированное количество, например, ровно 20 процессов, можно регулировать количество процессов динамически: например, от 5 до 30, в зависимости от количества запросов и нагрузки.

При этом по окончании обработки запроса процесс не умирает, а ждет следующего запроса, обрабатывает его и т.д.

Эту модель называют еще "pre-fork", и она заменила собой совсем старый подход, когда на каждый запрос происходил отдельный запуск программы-генератора страницы.

Таким образом работает большинство интерпретаторов с интерфейсом FastCGI: PHP/FastCGI, Perl/FastCGI, Ruby/FastCGI и другие. Сюда же относится WSGI(Python), да и вообще много чего.

Плюс этого способа - в том, что он простой. Еще бы - процесс является "средой в себе", со своими переменными, своим потоком управления, и никто случайно из одного процесса в другой не залезет.

Можно спокойно разобрать запрос, сгенерировать и выдать ответ посетителю.

А если вдруг где-то ошибка - так ничего: процессы изолированы на уровне операционной системы. Один упал зато другие живы.

Переход между запросами

Когда процесс заканчивает обработку запроса, в его адресном пространстве (где хранятся переменные, функции и т.п.) содержатся данные, которые были необходимы для генерации ответа.
Например, выбранные записи из базы данных, информация о посетителе, загруженные библиотеки и т.п.

Чтобы корректно обработать следующий запрос - нужно эти лишние данные удалить, почистить адресное пространство процесса. Для этого есть два основных подхода.

Интерпретатор чистит процесс

В случае, когда программа написана на каком-либо скриптовом языке, интерпретатор этого языка может очищать адресное пространство после окончания запроса.

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

Это происходит, например, в PHP и дает дополнительные гарантии от утечек памяти.

Самостоятельная очистка процесса

Чистка интерпретатором - решение, мягко говоря, жестковатое. Вместе с лишними переменными убивается и все полезное: библиотеки, синглтоны, вспомогательные объекты, которые можно было бы использовать при обработке следующего запроса.

Получается, что при новом запросе все эти объекты будут созданы заново. Но если мы используем большой мощный фреймворк, то каждый раз инициализовывать его заново - нерационально.

Поэтому естественное решение - возложить обязанности по чистке на программиста. Пусть сам удаляет то, что не нужно.

Типичный цикл обработки запроса при этом выглядит так:

// инициализация фреймворка
include MyFramework;

MyFramework::init();

// while ждет, пока процессу будет передан новый запрос
while(request = FastCGI::accept()) {
    MyFramework::process(request);
}

Функция MyFramework::process(request) читает нужные данные, генерирует и выдает ответ.
Объект запроса request является локальным для функции и по окончании ее работы будет удален, как и все остальные локальные и недостижимые снаружи объекты. А необходимые библиотеки и синглтоны - сохранены до следующего запроса.

Удобно? Удобно! Получается достаточно оптимальный цикл работы, который используется, например, в Perl/FastCGI, Ruby/FastCGI.

Общение между процессами

Общение между процессами и обмен данными, как правило, осуществляются через выделенную область памяти. При этом можно передавать только строковые представления объектов, которые необходимо превращать в реальный объект перед использованием.

Потоки OS

Потоки отличаются от процессов одной "мелочью" - общим адресным пространством.

Поэтому несколько одновременно пришедших запросов, хотя и будут обрабатываться разными потоками, имеют доступ к одним и тем же переменным.

При этом никто не гарантирует, что скорость обработки запросов будет одинаковая у разных потоков, и с этим связана одна важная проблема.

Например, есть код:

counter = 0
...
counter++
...
print counter

При поточном выполнении - вообще говоря, не факт что print counter выведет единицу. Может быть и ноль, если какой-то другой запрос начал обрабатываться и успел обнулить счетчик, пока мы были между "counter++" и "print counter". Переменная-то одна на всех.

Это и является основной сложностью поточного программирования - необходимость синхронизировать доступ к общим ресурсам между потоками: к соединениям с базой, переменным, и т.п.

Пример вверху решается очень легко: достаточно сделать counter локальной для функции обработки запроса:

process(request) {
  int counter = 0;
  ...
  counter++;
  ...
  print counter;
}

Каждый новый запрос вызовет process со своим объектом запроса, а локальные переменные - на то и локальные, чтобы быть видными только при текущем запуске функции. Так что в данном случае каждый запрос будет использовать свою собственную переменную counter и проблем не возникнет.

Кроме того, можно явно привязать переменную или объект к текущему потоку, использовать "критические" блоки кода, т.е. такие, в которые может зайти не более одного потока за раз, применять мьютексы и другие приемы многопоточного программирования.

Единое адресное пространство позволяет избежать затрат и сложностей, связанных с обменом данными между процессами. В частности, один поток может вызвать функцию другого и т.п.

Создание потока - более быстрая операция, чем создание нового процесса, что также является преимуществом многопоточной модели. Но аккуратное разделение и синхронизация переменных, отлов возникающих глюков могут сильно осложнить жизнь.

Такая модель - классическая для сервлет-контейнеров (Java): таких как Tomcat, Resin и другие. Она подробно описана в спецификации Servlet API 2.5. Она же используется и в C#.NET.

COMET - проблема процессов и потоков OS

COMET-приложение имеет другой режим функционирования, принципиально отличающийся от обычной генерации страницы.

1. Принять запрос (посетитель подключился к чату)
2. Ждать, пока появится сообщение
3. Получив сообщение, сгенерировать на его основе ответ
4. Выдать ответ
5. (Зависит от реализации) Закрыть соединение

Как правило, большую часть времени соединение висит в ожидании события. При этом выделенный процесс или даже поток на такое соединение - колоссальный перерасход ресурсов, прежде всего - памяти.

Одновременно подключенные к серверу 100 человек при классическом подходе дали бы 100 процессов(или потоков), большая часть которых будет висеть и ждать сообщения. Это достаточно большие затраты памяти, да и процессорное время тоже тратится, хотя и не в такой степени.

Проблема становится очевиднее при цифрах в 1000, 10.000 соединений, которые вполне достижимы в проектах, рассчитанных на большое количество одновременных пользователей: чаты, онлайн-игры, сайты знакомств и др.

Существует много COMET-серверов, но все они основаны на всего нескольких паттернах преодоления этой проблемы.

События в едином потоке (Twisted, node.js, ...)

Этот паттерн заключается в том, что все действия, которые требуют какого-либо ожидания или занимают существенное время, совершаются асинхронно.

На клиентской стороне такой паттерн вовсю используется в javascript:

var xhr = new XmlHttpRequest()
function processResult(result) {
 ...
}
xhr.onreadystatechange = processResult

xhr.send()
...

Вызов xhr.send() запускает выполнение запроса, и скрипт продолжает выполняться. А когда XmlHttpRequest выполнится, то он запустится обработчик события onreadystatechange - функция processResult.

Таким же образом устроено и серверное асинхронное программирование. Рассмотрим это на примере COMET-сервера, написанного на фреймворке Twisted (язык Python).

Пример на Twisted

Следующий пример иллюстрирует работу COMET-сервера в режиме long-poll. У него есть две точки подключения: /publish и /subscribe.

При подключении к /subscribe сервер держит соединение, пока не появится сообщение. Если подключилось много клиентов - все они будут висеть и ждать.

При подключении к /publish сервер считывает сообщение из параметра message и отправляет его всем клиентам, закрывая соответствующие соединения.

class ClientManager:
    clients = []
    def registerClient(self, client):
        self.clients.append(client)

    def broadcastMessage(self, message):
        for client in self.clients:
            client.write(message)
            client.finish()
        self.clients = []
clientManager = ClientManager()


class CometServer(resource.Resource):
    isLeaf = False
    def __init__(self):
        resource.Resource.__init__(self)
        self.registerChildren()

    def registerChildren(self):
        self.putChild('subscribe', Subscribe() )
        self.putChild('publish', Publish() )

    def getChild( self, path, req ):
        if path == '':
            return self
        else:
            return resource.Resource.getChild( self, path, req )


class Subscribe(resource.Resource):
    isLeaf = True
    def render_GET(self, req):
        clientManager.registerClient(req)
        return server.NOT_DONE_YET


class Publish(resource.Resource):
    isLeaf = True
    def render_POST(self, req):
        clientManager.broadcastMessage(req.args['message'][0])
        return "ok"


application = service.Application("App")
service = internet.TCPServer(8001, server.Site(CometServer()))
service.setServiceParent(application)

Разберем основные сущности подробнее:

ClientManager

Вспомогательный объект-синглтон, хранящий ожидающие сообщение соединения в списке clients.

Умеет добавлять подключившихся клиентов в список при вызове registerClient и рассылать всем из списка сообщение в broadcastMessage.

CometServer

Основной объект-синглтон, являющийся сервером. Он написан на фреймворке Twisted.web и делегирует обработку соединений объектам Subscribe() и Publish(), которые делают реальную работу.

Subscribe

Любой клиент. который делает запрос на адрес /subscribe, будет обработан (одним и тем же) объектом этого класса. В данном случае он добавляет клиента в список и возвращает специальное значение server.NOT_DONE_YET, получив которое, Twisted не закрывает соединение.

Таким образом, полученные соединения передаются для хранения в clientManager.

Publish

Получив POST-запрос, рассылает сообщение message его всем подписчикам.

код после Publish

Служебный код для инициализации и запуска серверного демона. Для нас он не важен.

Все происходит в едином адресном пространстве и в одном процессе/потоке.

Обработка множества одновременных соединений происходит благодаря асинхронному подходу.

Любое соединение, требущее ожидание, добавляется в список clientManager.clients, который используются при получении сообщений.

Что очень удобно, соединения доступны в любой момент. Их можно пронумеровать, положить в ассоциативный массив по сессии клиента и т.п.
Получив сообщение, можно выбрать нужные соединения из списка ожидающих и сделать с ними все что угодно.

Описанная асинхронная модель дает общее адресное пространство, и при этом - никаких проблем с синхронизацией доступа, свойственных потокам.

Блокирующееся API

Для того, чтобы работать в таком асинхронном режиме, код должен поддерживать соответствующее API и вызывать каллбэки. Не весь код сделан таким образом.

Например, ряд библиотек для работы с базами данных работает только синхронно, без каллбэков. Синхронное программирование, вообще, проще и привычнее большинству программистов.

Запускать такие функции в одном потоке нельзя - они заблокируют обработку новых соединений.
Чтобы все-таки сделать возможными такие синхронные вызовы, в Twisted есть прямой доступ (adbapi использует его неявно) к пулу потоков, размер которого можно указать явно вызовом:

reactor.suggestThreadPoolSize(100)

Функция, которая работает синхронно, может быть отправлена работать в отдельный поток вызовом deferToThread:

deferred = threads.deferToThread(syncSQL, "SELECT * FROM mytable")

deferred.addCallback(printResult)

Возвращаемое значение deferred является специальным объектом типа Deferred и вызовет функции, добавленные через addCallback, в момент получения результата.

В данном случае - в момент, когда функция из отдельного потока закончит выполнение.

Использовать потоки можно и в более явном виде, управлять ими вручную и т.п. Но, как правило, при асинхронном программировании все основные данные хранятся в единственном основном процессе, а потоки используются только для тех операций, которые не поддерживают асинхронный вызов.

Использование нескольких ядер/CPU

Основной процесс использует только одно ядро. Если нагрузка на процессор становится слишком большой, то никакого автоматического распараллеливания не происходит.

Поэтому используются два подхода.

• Первый - сложные операции выносятся в отдельные потоки. Если говорить о Python, то при этом возникают дополнительные сложности, связанные с Global Interpreter Lock, но они преодолимы.

  Разумеется, доступ к общим переменным и сетевые операции (они не thread-safe) потребуют дополнительной синхронизации.

• Второй - запускается несколько экземпляров сервера, и нагрузка балансируется между ними. В этом случае данные будут уже не в едином адресном пространстве.

  Но если у вас такое мощное приложение - возможно, это решение будет правильным, т.к. оно более масштабируемо: разные процессы проще будет разнести по разным машинам, чем потоки, использующие общее адресное пространство.

Пример на Node.js

Сервер Node.JS написан на Javascript и использует быструю библиотеку V8 для интерпретации Javascript-кода и libev для обработки событий. Он работает аналогичным образом. Все соединения обрабатываются единственным процессом/потоком.

Вот код такого же по функционалу COMET-сервера на Node.JS:

sys = require('sys'),
http = require('http');
url = require('url')

clientManager = new function() {
    var clients = []

    this.registerClient = function(client) {
      clients.push(client)
    }

    this.broadcastMessage = function(message) {
      for(var i=0; i<clients.length; i++) {
        var client = clients[i]
        client.writeHeader(200, {'Content-Type': 'text/plain;charset=utf-8'})
        client.write(message.toString(), 'utf-8')
        client.close()
      }
      clients = []
    }
}

http.createServer(function (req, res) {
  var urlParsed = url.parse(req.url, true)

  if (urlParsed.pathname == '/publish') {
    clientManager.broadcastMessage(urlParsed.query.message)
    res.writeHeader(200, {'Content-Type': 'text/plain;charset=utf-8'})
    res.write('ok', 'utf-8')
    res.close()
  }

  if (urlParsed.pathname == '/subscribe') {
    clientManager.registerClient(res)
  }

}).listen(8000);

Как видно, примеры на Node.JS и Twisted очень похожи. Одинаковой функциональностью обладает clientManager. Различие только в синтаксисе.

Конечно Node.JS является гораздо более молодым фреймворком и функционально далек от такого гиганта как Twisted, но он легче, написан на новых библиотеках и вполне юзабелен для несложных приложений.

Continuations (Jetty, GlassFish...)

Этот паттерн прибрел широкую известность благодаря веб-серверу Jetty, написанному на Java.

Он заключается в том, что запрос посетителя "упаковывается" в специальный объект Continuation, который передается обработчику асинхронного события.

Когда событие происходит, обработчик либо выдает ответ самостоятельно, либо вызывает метод continuation.resume(), который заставляет сервер обработать запрос заново, но при этом объект continuation будет содержать данные события.

Вот так это выглядит:

public class SubscribeServlet extends HttpServlet {
    
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        
        Continuation continuation = ContinuationSupport.getContinuation(req);

        if (continuation.isInitial()) {
            continuation.suspend();
            ClientManager.getInstance().registerClient(continuation);
        } else {
            String message = (String)continuation.getAttribute("message");
            resp.setContentType("text/plain;charset=utf-8");
            resp.getWriter().print(message);            
        }

    }
}

Рассмотрим происходящее в деталях:

1. Пришел запрос в SubscribeServlet
2. Запрос оборачивается в объект continuation.
3. При создании continuation.isInitial() возвращает true, поэтому будет исполнен код:
   

   continuation.suspend();
   ClientManager.getInstance().registerClient(continuation);
   

   Первая строка говорит серверу, что объект continuation необходимо оставить в памяти после завершения запроса и не закрывать соединение, а вторая - регистрирует его среди ожидающих клиентов.

4. На этом поток заканчивает обработку запроса и может быть использован для обработки новых запросов.

В дальнейшем приходит сообщение на /publish. Такие запросы попадают в PublishServlet:

public class PublishServlet extends HttpServlet {
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        ClientManager.getInstance().broadcastMessage(req.getParameter("message"));
        resp.setContentType("text/plain;charset=utf-8");
        resp.getWriter().print("ok");
    }
}

Этот сервлет вызывает метод ClientManager#broadcastMessage, в котором и происходит все интересное:

public class ClientManager {
    ArrayList<Continuation> continuations = new ArrayList<Continuation>();

    public synchronized void registerClient(Continuation continuation) {
        continuations.add(continuation);
    }

    public synchronized void broadcastMessage(String message) {
        for(Continuation continuation: continuations) {
            continuation.setAttribute("message", message);
            continuation.resume();
        }
        continuations.clear();
    }

    // синглтон

    private static ClientManager ourInstance = new ClientManager();

    public static ClientManager getInstance() {
        return ourInstance;
    }

    private ClientManager() {   }

}

Как видно из последних строк кода, этот объект является синглтоном.

Список ожидающих запросов, обернутых в объект типа Continuation, хранится в свойстве continuations. Это свойство является обычным ArrayList, операции с которым не thread-safe.

При вызове broadcastMessage во все continuations записывается поступившее сообщение (туда можно записать любой объект) и производится вызов continuation.resume().

При таком вызове сервер запускает обработку запроса, на основе которого создан continuation, заново. При этом continuation.isInitial() уже вернет false, т.к. запрос не является изначальным.

В SubscribeServlet это приведет к выполнению нижней ветки кода:

String message = (String)continuation.getAttribute("message");
resp.setContentType("text/plain;charset=utf-8");
resp.getWriter().print(message);

Здесь мы не вызываем continuation.suspend(), поэтому соединение будет закрыто.

Обратим внимание, что методы синглтона clientManager являютя критическими, т.е. помечены как synchronized. Это обязательно, иначе могут возникнуть ошибки одновременного доступа.

Альтернатива

Альтернативная модель использования Continuations заключается в генерации ответа непосредственно при обработке события, без пробрасывания запроса в новый цикл обработки.

В примере выше это означало бы, что вместо вызова continuation.resume() метод broadcastMessage сгенерировал бы ответ, отправил его и закрыл continuation.

Это кажется более простым вариантом, но вспомним, что клиенты обрабатываются в цикле:

for(Continuation continuation: continuations) {
            continuation.setAttribute("message", message);
            continuation.resume();
        }

Если вместо continuation.resume(); вставить сколько-нибудь сложную генерацию ответа, то придется самостоятельно заботиться о том, чтобы она была асинхронной, чтобы цикл не обрабатывал клиентов по очереди - а их может быть много.

То есть, придется самостоятельно работать с потоками и т.п.
С другой стороны, для генерации простого ответа подойдет и такой способ: в нем отсутствуют излишние затраты на повторный вход в поток обработки запроса.

Обе модели также реализованы в сервере GlassFish и учтены в спецификации Servlet API 3.0.

Использование нескольких ядер/CPU

При описанном подходе в основе всего лежит ThreadPool. Потоки запускаются, обрабатывают запрос и, по окончании обработки, возвращаются в пул и могут быть использованы заново.

При этом соединение инкапсулируется в объекте типа Сontinuation и хранится в памяти. Вызов этого объекта позволяет продолжить работу с отложенным соединением в любой момент.

Использование потоков естественным образом приводит к загрузке всех доступных ядер/CPU.

Микронити (Erlang, Stackless Python...)

Предыдущие методы (асинхронная обработка и continuations) объединяет общий подход к задаче обработки множества соединений.

Они сохраняют состояние в памяти и освобождают поток для дальнейшей работы.

"Микронити" - это решение с совершенно другой стороны. Вместо того, чтобы освобождать поток, он делается максимально легким, чтобы можно было поддерживать 100, 1000, 10.000 и более потоков без существенной загрузки ресурсов.

Это достигается в первую очередь благодаря отказу от потоков OS и от использования стека операционной системы. Вместо этого виртуальная машина интерпретатора поддерживает собственный стек и механизмы переключения между контекстами.

Одним из самых известных языков, основанных на микронитях, является Erlang.

Он является функциональным языком программирования со специфическим синтаксисом и множеством оригинальных возможностей, включая встроенную поддержку кластеризации, базы данных, горячую замену кода без остановки обслуживания и другие.

Для реализации COMET-сервера я использовал фреймворк MochiWeb.

Пример на Mochiweb(Erlang)

Вот как он выглядит:

room(Users) ->
    receive
        {From, subscribe} ->
            From ! subscribed,
            room([From | Users]);
        {From, post, Message} ->
            From ! posted,
            lists:foreach(fun(User) ->
                    % broadcast the message
                    User ! Message
                end, Users),
            room([]);
        _Any ->
            room(Users)
    end.

get_the_room() ->
    % does the room exists?
    Pid = whereis(theroom),
    if
        is_pid(Pid) ->
            % yup
            Pid;
        true ->
            % create it
            NewPid = spawn(fun() ->
                room([])
            end),
            register(theroom, NewPid),
            NewPid
    end.


loop(Req, DocRoot) ->
    "/" ++ Path = Req:get(path),
    case Req:get(method) of
        Method when Method =:= 'GET'; Method =:= 'HEAD' ->
            case Path of
                "subscribe" ->
                    Room = get_the_room(),
                    Room ! {self(), subscribe},
                    receive
                        subscribed ->
                            % subscription is ok
                            % now wait for a message
                            receive
                                Message ->
                                    ok
                            after ?TIMEOUT ->
                                % we waited too long
                                Message} = <<"timeout">>
                            end
                    after 50000 ->
                        % subscription failed on time
                        Message = <<"timeout">>
                    end,

                    % send back the JSON message
                    Req:ok({"text/plain;charset=utf-8", Message});
                end;
        'POST' ->
            case Path of
                "publish" ->
                    Data = Req:parse_post(),
                    Room = get_the_room(),
                    % post
                    Room ! {self(), post, list_to_binary(proplists:get_value("message", Data))},
                    receive
                        posted ->
                            % posted
                            ok
                    after 50000 ->
                        % something went wrong
                        notok
                    end,

                    % send back the message
                    Req:ok({"text/plain", "ok"});
                _ ->
                    Req:not_found()
            end;
        _ ->
            Req:respond({501, [], []})
    end.

На тот случай, если Эрланг для вас внове - позвольте объяснить происходящее.

Каждая функция в Erlang производит вычисления выражений, разделенных запятыми и возвращает последнее вычисленное.

Итак, рассмотрим код выше по частям.

get_the_room

Все подключения хранятся в процессе с именем theroom. Это аналог рассмотренного выше clientManager, но вместо синглтона используется отдельный процесс.

Метод get_the_room пытается получить Pid процесса с именем theroom и, если такого процесса нет, генерирует его вызовом встроенного метода spawn.

get_the_room() ->
    % does the room exists?
    Pid = whereis(theroom),
    if
        is_pid(Pid) ->
            % yup
            Pid;
        true ->
            % create it
            NewPid = spawn(fun() ->
                room([])
            end),
            register(theroom, NewPid),
            NewPid
    end.

Единственная задача этого метода - запуск процесса theroom. При запуске процесса ему передается для выполнения функция room.

room

Первый аргумент room - текущий список подключенных посетителей. Изначально при вызове spawn этот список пустой.

room(Users) ->
    *!*receive*/!*
        {From, subscribe} ->
            From ! subscribed,
            room([From | Users]);
        {From, post, Message} ->
            From ! posted,
            lists:foreach(fun(User) ->
                    % broadcast the message
                    User ! Message
                end, Users),
            room([]);
        _Any ->
            room(Users)
    end.

Директива receive заставляет процесс висеть, ожидая сообщение.

При получении сообщения subscribe от процесса From, она отправляет в ответ сообщение subscribed:

% этот синтаксис означает отправку сообщения subscribed 
% процессу с идентификатором From
From ! subscribed,

Сообщение отправляется асинхронно, т.е. функция продолжает работу, вызывая room([From | Users]) - то есть саму себя, но со списком, в который добавлен процесс From (операция | означает добавление элемента в список).

Этот рекурсивный запуск приведет к новому вызову receive и так далее. Обычное функционирование процесса, только вместо цикла - рекурсивный вызов.

Ветка {From, post, Message} активируется при получении сообщения post и рассылает сообщение всем процессам из текущего списка.

From ! posted,
lists:foreach(fun(User) ->
    % Отослать сообщение Message процессу User
     User ! Message
end, Users),

Как видите, каждому посетителю соответствует отдельный процесс.

После рассылки сообщений рекурсивный вызов room([]) происходит с пустым списком, и процесс продолжает работу.

loop(Req, DocRoot)

Эта функция обрабатывает входящие запросы. Как легко видеть из кода, она разбрасывает их по веткам subscribe и publish.

Ветка subscribe:

% получить идентификатор процесса комнаты
Room = get_the_room(),
% отослать сообщение subscribe (асинхронно)
Room ! {self(), subscribe},
% подвиснуть в ожидании ответа от комнаты
receive
    subscribed ->
        % подписка удалась, подождать сообщения
        receive
            Message ->
                ok
        % таймаут ожидания
        after ?TIMEOUT ->
            % слишком долго ждали сообщения
            Message} = <<"timeout">>
        end
after 50000 ->
    % не сигнала об успешной подписке
    Message = <<"timeout">>
end,
% отправить ответ или ошибку посетителю
Req:ok({"text/plain;charset=utf-8", Message});

Ветка publish работает аналогичным образом, разве что попроще.

Благодаря использованию легких процессов, 100, 1000 и даже 100.000 процессов легко обрабатываются виртуальной машиной Erlang/OTP.

Erlang + Continuations

Концепция микропотоков никак не противоречит Continuations.
Процессы эрланга могут быть заморожены (hibernate), при этом максимально очищается связанная с процессом память и освобождаются ресурсы, а затем разморожены при получении сообщения.

Это особенно удобно, когда процессы висят подолгу, ожидая сообщение.

С этим связаны некоторые сложности в программировании, т.к. hibernate, кроме всего прочего, очищает стек, но они вполне преодолимы.

Как вы, наверное, уже догадались(или знали), Erlang позволяет создавать ну очень производительные, масштабируемые и надежные приложения.

Тюнинг OS

Для работы с большим количеством соединений необходимы некоторые настройки операционной системы.

Для Linux это можно сделать следующими строчками в /etc/sysctl.conf:

# General gigabit tuning:
net.core.rmem_max = 16777216
net.core.wmem_max = 16777216
net.ipv4.tcp_rmem = 4096 87380 16777216
net.ipv4.tcp_wmem = 4096 65536 16777216
net.ipv4.tcp_syncookies = 1

# this gives the kernel more memory for tcp
# which you need with many (100k+) open socket connections
net.ipv4.tcp_mem = 50576 64768 98152
net.core.netdev_max_backlog = 2500
net.ipv4.tcp_keepalive_time = 300
net.core.somaxconn = 4096

Плюс - поднять лимит на количество открытых файлов. Например, только для root: (/etc/security/limits.conf)

# ....
*               soft    nofile          1024
*               hard    nofile          2048
root            soft    nofile          32768
root            hard    nofile          32768

Исходники

Исходники всех серверов, упомянутых в данной статье, вы можете скачать.

• Скачать исходники серверов на Jetty / Node.JS / Twisted / Mochiweb

Открытые COMET-сервера

Ряд серверов предлагает дополнительные обертки и фреймворки, упрощающие COMET, включая javascript-библиотеки и различные режимы работы: polling / streaming / long-poll.

Вот список самых распространенных открытых серверов:

Java

• Glassfish с компонентом Grizzly Comet
• CometD

Javascript

• Node.JS

Python

• Orbited
• Tornado

Erlang

• Mochiweb

Ruby

• Juggernaut (for Rails)

Java - сервера используют Continuations, Python/Ruby/Javascript - события в едином потоке, Erlang-реализация основана на микронитях с возможностью hibernate.

Соответственно, в плюсе у Java-серверов - естественная масштабируемость на множество потоков и использование множества CPU, у Python/Ruby/Javascript - асинхронное программирование в едином потоке без синхронизаций, Erlang обеспечивает кластеризацию и отказоустойчивость на множестве серверов, очень хорошие характеристики по производительности и затратам памяти.

Вам выбирать, что подойдет для ваших приложений.