2021-01-04

F# WebSocket Server

Аннотация

Идея писать веб-сервера и веб-фреймворки на всех языках у меня возникла с тех пор, когда я понял, что то, что я сделал для экосистемы Erlang: направление фреймворков для предприятий под общим брендом N2O, а сейчас как часть платформы erp.uno; вполне применимо и для других языков и платформ. В этой статье представлена версия вебсокет-сервера для языка программирования F# — ws.erp.uno. Адрес пакета: nuget.org/packages/ws. Адрес репозитория: erpuno/ws.

Предисловие

Haskell. Первый эксперимент был совершен Андреем Мельниковым в виде порта для Хаскеля: N2O.HS, позже более полную версию с N2O и NITRO с экзестенциальными сигнатурами сделал Марат Хафизов, который управляет Github организацией O3 и сайтом o3.click. Мне совершенно непонятно, почему ни один хаскель программист, которые вроде как должны восхищаться минимализмом, не идет по этому пути, а обычно ищет правды в таких фреймворках как UrWeb, IHP, UnisonWeb. На мой взгляд — всё это переусложненные штуки.

Standard ML. Также в академических целях, Марат Хафизов сделал порт связки веб-сервера N2O и веб-фреймворка NITRO на язык Standard ML (обе главные версии SML/NJ и MLton) — эта работа представлена Github организацией O1. Это тот язык, который я считаю уместно преподавать как первый академический язык программирования (до знакомства с промышленными языками Erlang, F#, Haskell).

Lean. Для закрепления своей идеи и более четкой и точной ее артикуляции я попросил Siegmentation Fault сделать порт на еще более формальный язык программирования, математический прувер Lean 4. Эта версия связки веб-сервера N2O и веб-фреймворка NITRO представлена Github организацией O89 и сразу двумя сайтами: lean4.dev и bum.pm. Последний представляет собой пакетный менеджер написанный на Lean 4, который нам помогает администрировать Александр Темерев из CERN. Lean 4 N2O проекты залайкал Леонардо де Мура, автор Lean и Z3, чему мы безмерно рады.

Идиоматический веб-сервер на F#

Критерии идиоматичности могут каждым восприниматься по разному, но в основном это означаем минимум прелюдий и максимум сути, так или иначе основная мантра всех минималистов в общем и N2O инфраструктуры в частности. Так в современные критерии идиоматичности веб-сервера для языка F# я бы выделил следующее: 1) использование системных классов System.Net.WebSockets, которые уже предоставляют буферизированные енкодер и декодер фреймов стандарта RFC 6455; 2) сервер должен быть построен на Async компютейшинал экспрешинах; 3) для управления асинхронными потоками выполнения должен использоваться MailboxProcessor, а не самописная система воркеров, которая хоть и поможет выжать последнее из F# (у меня получилось 14 миллионов сообщений в секунду), но не продемонстрирует сути, так как будет девиацией в сторону рантаймов; 4) Использование классов TcpListener и TcpClient, NetworkStream. Больше ни чем не разрешается пользоваться!

Что почитать перед написанием?

Немного погуглив, я понял что интернету нехватает статьи, которая описывает историю понятия асинхронных вычислений и вычислительных выражений, которые в народе известны по ключевым словам async/await. Вижу статью, которая называется "Survey of Brief Async history", в которой будет показана ретроспектива Async технологии:

0) J operator 1965;
1) LISP call/cc 1968;
2) Erlang 1986;
3) Concurrent ML 1998;
4) Haskell async 2004;
5) C# async yield 2006;
6) Perl IO:Async 2007;
7) F# Async 2010
8) C#/PHP Async 2012
9) Python async 2015
10) ECMAScript async 2017

Основополагающей статьей по F# async я бы назвал F# Async Guide Лео Городинского, jet.com. Основной книгой, которую я бы порекомендовал полистать перед знакомством с F# — это "Expert F# 4.0" автора языка Дона Сайма. Основной презентацией по F# Async я бы назвал доклад Дона Сайма на митапе в Лондоне — Some F# for the Erlang programmer. Вооружившись этими документами и этим Gist сниппетом я ухал во Львов писать самый идиоматичный вебсокет-сервер.

Витрина

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


  open N2O

  module Program =

    [<EntryPoint>]
    let main _ =
        let mutable ret = 0

        try Stream.protocol <- fun x -> x
            use ws = Server.start "0.0.0.0" 1900
            System.Threading.Thread.Sleep -1
        with exn ->
            printfn "EXIT: %s" exn.Message
            ret <- 1
        ret

Архитектура асинхронных процессов

Для тех, кто знаком с архитектурой Erlang/OTP, известно, что проектирование сетевых приложений начинается с дерева супервижина легковесных процессов и протоколов которые определяют их взаимодействие. Подчиненные дочерние процессы обычно разделяют токены времени жизни CancellationToken, благодаря чему исключения возникшие в родительских процессах могут отменить все дерево подпроцессов. Поэтому в циклах процессов присутствует выражение


    while not ct.IsCancellationRequested do

Наш вебсокет-сервер состоит из 7 асинхронных процессов:


         [Sup]    [L]*
        /        /
 [start]--[S]--[C]*
        \        \
         [H]      [T]*

Легенда этого дерева такова: [start] нода представляет собой точку входа, из которой будут рождаться остальные асинхронные процессы, соотвествует функции start; [S] нода соовтествует асинхронному процессу, который представлен функцией listen; [Sup] нода соотвествует функции startSupervisor; [C] нода соотвествует функции startClient; [H] нода соотвествует функции heartbeat; [L] нода соотвествует функции loop; [T] нода соотвествует функции telemetry. Звездочкой будем обозначать процессы, количество которых зависит от количества активных соединений: [C]*, [L]*, [T]*.

Протоколы взаимодействия

В момент рождения клиента [C] в родительском процессе сервера [S] происходит нотификация [S]->[Sup] по так называемому протоколу супервижина Sup с одноименным типом. Публичный протокол публичной функции Stream.protocol представлен типом Msg, который предназначен для управления асинхронным процессом [L].

Система серверных пингов реализована совместимой с протоколами Sup и Msg, хартбит процесс [H] через интервал времени посылает Tick сообщение в супервизор [Sup], который в свою очередь шлет броадкаст для всех клиентов телеметрии [T] созданных на той же очереди, что и [C], т.е. тот же протокол.

Процесы [T], [L] и [C] разделяют WebSocket стрим и все связаны с супервизором сервера, нотифицируя его в случае возникновения исключений.


    type Msg =
        | Bin of byte array
        | Text of string
        | Nope

    type Sup =
        | Connect of MailboxProcessor<Msg> * WebSocket
        | Disconnect of MailboxProcessor<Msg>
        | Close of WebSocket
        | Tick

RFC 6455 Хендшейк

Функции обработки HTTP хедеров. isWebSocketsUpgrade ищет в хедерах пару Upgrade и WebSocket. getLines возвращает хедеры как массив строк, а функция getKey возвращает значение хедера по его ключу.


    let isWebSocketsUpgrade (lines: string array) =
        Array.exists (fun (x:string) -> "upgrade: websocket" = x.ToLower()) lines

    let getKey (key: string) arr =
        try let f (s: String) = s.StartsWith(key)
            (Array.find f arr).[key.Length + 1..]
        with _ -> ""

    let getLines (bytes: byte array) len =
        if len > 8 then
            bytes.[..(len - 9)]
            |> UTF8Encoding.UTF8.GetString
            |> fun hs -> hs.Split([| "\r\n" |],
                   StringSplitOptions.RemoveEmptyEntries)
        else
            [||]

Функция RFC 6455 ответа называется handshake. Этой функциональности насколько мне известно нет в системных неймспейсах.


    let acceptString6455 acceptCode =
        "HTTP/1.1 101 Switching Protocols\r\n" +
        "Upgrade: websocket\r\n" +
        "Connection: Upgrade\r\n" +
        "Sec-WebSocket-Accept: " + acceptCode + "\r\n\r\n"

    let handshake lines =
        (getKey "Sec-WebSocket-Key:" lines) + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
        |> Encoding.ASCII.GetBytes
        |> SHA1CryptoServiceProvider.Create().ComputeHash
        |> Convert.ToBase64String
        |> acceptString6455
        |> Encoding.ASCII.GetBytes

Асинхронные процессы сервера

Первый процесс [start], представляет собой точку входа, где стартует сразу три процесса: процесс супервизора всех соединений [Sup], процесс сервера слушателя соединений [S] и, если включен флаг Server.ticker, процесс сердцебиений, который работает как интервальный циклический таймер [H]. Эпилог процесса [start] содержит кенселяцию токена глобального для всех подпроцессов при освобождении переменной, которая соджержит вебсокет-сервер во внешнем коде.


    let start (addr: string) (port: int) =
        let cts = new CancellationTokenSource()
        let token = cts.Token
        let sup = startSupervisor token
        let listener = TcpListener(IPAddress.Parse(addr), port)

        try
            listener.Start(10)
        with
            | :? SocketException ->
                failwithf "%s:%i is using by another program" addr port
            | err ->
                failwithf "%s" err.Message

        Async.StartImmediate(listen listener token sup, token)
        if ticker then Async.StartImmediate(heartbeat interval token sup, token)

        { new IDisposable with
            member x.Dispose() = cts.Cancel() }

Второй процесс [Sup], супервизор является чистой функцией, которая обрабатывет сообщения супервижин протокола о регистрации и смерти новых соединений. Тут же происходит бродкаст сообщений как реакция на сердцебиение тикера.


    let startSupervisor (ct: CancellationToken) =
        MailboxProcessor.Start(
            (fun (inbox: MailboxProcessor<Sup>) ->
                let listeners = ResizeArray<_>()
                async {
                    while not ct.IsCancellationRequested do
                        match! inbox.Receive() with
                        | Close ws -> ()
                        | Connect (l, ns) -> listeners.Add(l)
                        | Disconnect l -> listeners.Remove(l) |> ignore
                        | Tick -> listeners.ForEach(fun l -> l.Post Nope)
                }),
            cancellationToken = ct
        )

Интервальный таймер сердцебиения [H].


    let heartbeat (interval: int)
                  (ct: CancellationToken)
                  (sup: MailboxProcessor<Sup>) =
        async {
            while not ct.IsCancellationRequested do
                do! Async.Sleep interval
                sup.Post(Tick)
        }

Главный цикл процесса [S], который принимает новые TCP соединения и стартует новых клиентов [C].


    let listen (listener: TcpListener)
               (ct: CancellationToken)
               (sup: MailboxProcessor<Sup>) =
        async {
            while not ct.IsCancellationRequested do
                let! client = listener.AcceptTcpClientAsync() |> Async.AwaitTask

                client.NoDelay <- true
                startClient client sup ct |> ignore
        }

Асихронный процесс [C] с очередью (MailboxProcessor) обработки TCP соединений или, проще говоря, TCP клиент. Это точка входа для клиентского соединения, именно здесь происходит хендшейк. В случае успешного хендшейка мы шлем RFC 6455 ответ и запускаем сразу два асинхронных процесса: первый это сам цикл обработки вебсокет сообщений [L], а также, если установлен флаг Server.ticker мы запускаем процесс телеметрии [T], который разделяет WebSocket стрим и может осуществлять туда асинхронный сброс сообщений, конкурируя с основным циклом [L]. Такие процессы существуют всегда в паре.


    let startClient (tcp: TcpClient)
                    (sup: MailboxProcessor<Sup>)
                    (ct: CancellationToken) =
        MailboxProcessor.Start(
            (fun (inbox: MailboxProcessor<Msg>) ->
                async {
                    let ns    = tcp.GetStream()
                    let size  = tcp.ReceiveBufferSize
                    let bytes = Array.create size (byte 0)
                    let! len  = ns.ReadAsync(bytes, 0, bytes.Length)
                                |> Async.AwaitTask
                    let lines = getLines bytes len
                    match isWebSocketsUpgrade lines with
                    | true ->
                        do! ns.AsyncWrite (handshake lines)
                        let ws =
                            WebSocket.CreateFromStream(
                                (ns :> Stream), true, "n2o", TimeSpan(1, 0, 0))
                        sup.Post(Connect(inbox, ws))
                        if ticker then Async.Start(telemetry ws inbox ct sup, ct)
                        return! looper ws size ct sup
                    | _ -> tcp.Close()
                }),
            cancellationToken = ct
        )

Процесс телеметрии [T] слушает очередь, и на любое сообщение, шлет в вебсокет канал текстовое сообщение "TICK".


    let telemetry (ws: WebSocket)
                  (inbox: MailboxProcessor<Msg>)
                  (ct: CancellationToken)
                  (sup: MailboxProcessor<Sup>) =
        async {
            try
                while not ct.IsCancellationRequested do
                    let! _ = inbox.Receive()
                    do! send ws ct (Text "TICK")
            finally
                sup.Post(Disconnect <| inbox)

                ws.CloseAsync(WebSocketCloseStatus.PolicyViolation, "TELEMETRY", ct)
                |> ignore
        }

Главный цикл обработки сообщений [L], в котором происходит создание буферизированого WebSocket стрима, тип которого явно присутствует в протоколе супервижина. Также здесь выделяется глобальный для всего цикла буфер, куда копируются байты их сокета с помощью ReceiveAsync. При возникновении исключения происходит нотификация супервизора с помощью сообщения Close, которое сигнализирует о разрыве соединения, например в случае ошибки валидации UTF-8.


    let looper (ws: WebSocket)
               (bufferSize: int)
               (ct: CancellationToken)
               (sup: MailboxProcessor<Sup>) =
        async {
            try
                let mutable bytes = Array.create bufferSize (byte 0)
                while not ct.IsCancellationRequested do
                    let! result =
                        ws.ReceiveAsync(ArraySegment<byte>(bytes), ct)
                        |> Async.AwaitTask

                    let recv = bytes.[0..result.Count - 1]

                    match (result.MessageType) with
                    | WebSocketMessageType.Text ->
                        do! protocol (Text (Encoding.UTF8.GetString recv))
                            |> send ws ct
                    | WebSocketMessageType.Binary ->
                        do! protocol (Bin recv)
                            |> send ws ct
                    | WebSocketMessageType.Close -> ()
                    | _ -> printfn "PROTOCOL VIOLATION"
            finally
                sup.Post(Close <| ws)

                ws.CloseAsync(WebSocketCloseStatus.PolicyViolation, "LOOPER", ct)
                |> ignore
        }

Функции терминации канала наследуюет архаическое на мой взгляд разделение текстовых и бинарных сообщений. Как показывает практика трактовка всего как бинарных сообщений только улучшает семантику протокола.


    let sendBytes (ws: WebSocket) ct bytes =
        ws.SendAsync(ArraySegment<byte>(bytes),
            WebSocketMessageType.Binary, true, ct) |> ignore

    let send ws ct (msg: Msg) =
        async {
            match msg with
            | Text text -> sendBytes ws ct (Encoding.UTF8.GetBytes text)
            | Bin arr -> sendBytes ws ct arr
            | Nope -> ()
        }

Что дальше?

Дальше идут три фазы:

1) Контекст пира: порт, айпишник, хедеры, эндпойнт, служебная информация;
2) BERT сериализация для совместимости с клиентской инфраструктурой N2O;
3) Имплементация NITRO протокола.

Благодарности

Хочется поблагодарить всех, кто ставил лайки нашему проекту, а осоебнно Филлипа Картера, програмного менеджера .NET и F# ❤ Мы чрезвычайно воодушевлены!

Авторы

Максим Сохацкий, Игорь Городецкий, Siegmentation Fault