DEV Community: Frevib

A TCP server with kqueue

Frevib — Wed, 25 Sep 2019 18:21:18 +0000

Kernel queue (kqueue)

If you are programming with asynchronous (non-blocking) frameworks, then most likely this framework is using kernel polling APIs under the hood. Using an infinite loop, kernel polling APIs check for events and notify the caller when an event has occurred. For example when an HTTP request is fired, in each iteration of the loop a check is done to see if data already has returned from the server. The major OSes have implemented kernel polling in a different way:

Linux uses epoll
BSD and OSX use kqueue
Windows uses IOCP

epoll and kqueue are functionally almost the same, but there are some API differences. Let's write a bare minimum TCP server to understand how kqueue works.

Steps for creating a kqueue TCP server

Create a listening socket. This includes creating the socket with socket(2) and binding the socket to and IP and port with bind(2).
Create kevents we want to monitor with the helper macro EV_SET. Then create the kqueue itself with kqueue(2), and add the kevents to the kqueue using kevent(2).
Create an "event loop". This is just an infinite loop that checks if new clients have connected to our listening socket, and creates a new socket for the actual connection to the client using accept(8).

Code including Makefile on github

#include <stdio.h>
#include <stdlib.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/event.h>
#include <string.h>
#include <unistd.h>

int main()
{
    // All needed variables.
    int socket_listen_fd,
        portno = 1815,
        client_len,
        socket_connection_fd,
        kq,
        new_events;
    struct kevent change_event[4],
        event[4];
    struct sockaddr_in serv_addr,
        client_addr;

    // Create socket.
    if (((socket_listen_fd = socket(AF_INET, SOCK_STREAM, 0)) < 0))
    {
        perror("ERROR opening socket");
        exit(1);
    }

    // Create socket structure and bind to ip address.
    bzero((char *)&serv_addr, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = INADDR_ANY;
    serv_addr.sin_port = htons(portno);

    if (bind(socket_listen_fd, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) < 0)
    {
        perror("Error binding socket");
        exit(1);
    }

    // Start listening.
    listen(socket_listen_fd, 3);
    client_len = sizeof(client_addr);

    // Prepare the kqueue.
    kq = kqueue();

    // Create event 'filter', these are the events we want to monitor.
    // Here we want to monitor: socket_listen_fd, for the events: EVFILT_READ 
    // (when there is data to be read on the socket), and perform the following
    // actions on this kevent: EV_ADD and EV_ENABLE (add the event to the kqueue 
    // and enable it).
    EV_SET(change_event, socket_listen_fd, EVFILT_READ, EV_ADD | EV_ENABLE, 0, 0, 0);

    // Register kevent with the kqueue.
    if (kevent(kq, change_event, 1, NULL, 0, NULL) == -1)
    {
        perror("kevent");
        exit(1);
    }

    // Actual event loop.
    for (;;)
    {
        // Check for new events, but do not register new events with
        // the kqueue. Hence the 2nd and 3rd arguments are NULL, 0.
        // Only handle 1 new event per iteration in the loop; 5th
        // argument is 1.
        new_events = kevent(kq, NULL, 0, event, 1, NULL);
        if (new_events == -1)
        {
            perror("kevent");
            exit(1);
        }

        for (int i = 0; new_events > i; i++)
        {
            int event_fd = event[i].ident;

            // When the client disconnects an EOF is sent. By closing the file
            // descriptor the event is automatically removed from the kqueue.
            if (event[i].flags & EV_EOF)
            {
                printf("Client has disconnected");
                close(event_fd);
            }
            // If the new event's file descriptor is the same as the listening
            // socket's file descriptor, we are sure that a new client wants 
            // to connect to our socket.
            else if (event_fd == socket_listen_fd)
            {
                // Incoming socket connection on the listening socket.
                // Create a new socket for the actual connection to client.
                socket_connection_fd = accept(event_fd, (struct sockaddr *)&client_addr, (socklen_t *)&client_len);
                if (socket_connection_fd == -1)
                {
                    perror("Accept socket error");
                }

                // Put this new socket connection also as a 'filter' event
                // to watch in kqueue, so we can now watch for events on this
                // new socket.
                EV_SET(change_event, socket_connection_fd, EVFILT_READ, EV_ADD, 0, 0, NULL);
                if (kevent(kq, change_event, 1, NULL, 0, NULL) < 0)
                {
                    perror("kevent error");
                }
            }

            else if (event[i].filter & EVFILT_READ)
            {
                // Read bytes from socket
                char buf[1024];
                size_t bytes_read = recv(event_fd, buf, sizeof(buf), 0);
                printf("read %zu bytes\n", bytes_read);
            }
        }
    }

    return 0;
}

Kotlin extension function vs function literal with receiver

Frevib — Thu, 22 Aug 2019 09:02:30 +0000

Kotlin extension function and function literal with receiver

In Kotlin it is possible to add a method (called member function in Kotlin) to an existing class. This is called an extension function.

It is also possible to access a member function from a class inside a function literal. This is called a function literal with receiver.

Extension function

Say you want to add a new member function to the StringBuilder class, e.g. appendMonkey() that appends the string “monkey” to a string. Because StringBuilder is in the Java SDK, we cannot modify it. In Kotlin we can define an extension function that extends an existing class like StringBuilder with a new member function.

We define the appendMonkey() extension function as follows:

fun StringBuilder.appendMonkey(): StringBuilder = this.append("monkey")

this refers to the StringBuilder object, and we can omit it:

fun StringBuilder.appendMonkey(): StringBuilder = append("monkey")

Full example:

import java.lang.StringBuilder

fun main() {
    val sb = StringBuilder()

    sb.append("Hello extension function ")
    sb.appendMonkey()

    println(sb.toString())
}

fun StringBuilder.appendMonkey(): StringBuilder = append("monkey")

Function literal with receiver

Closely related to the extension function is the function literal with receiver. There are two types of function literals:

lambda
anonymous function

In practice you will most likely use the lambda expression, so we use that in the example below.

Where with extension functions you can add a new member function to an existing class, with a function literal with receiver you can access the member functions of an existing class inside the lambda block (inside the curly braces {}).

Let's create a function literal with receiver that adds the string "monkey" to a string using the StringBuilder class.

val lambdaAppendMonkey: StringBuilder.() -> StringBuilder = { this.append("monkey") }

Again, this refers to the StringBuilder object, so we omit it:

val lambdaAppendMonkey: StringBuilder.() -> StringBuilder = { append("monkey") }

Full example:

 import java.lang.StringBuilder

fun main() {
    val lambdaAppendMonkey: StringBuilder.() -> StringBuilder = { append("monkey") }

    val sb = StringBuilder()
    sb.append("Hello lambda ")

    println(lambdaAppendMonkey(sb).toString())
}

Here, inside the block of the lambda we can access the append() member function because we explicitly stated the return type to be StringBuilder.() -> StringBuilder. This return type basically says:

"return a function that takes no arguments and returns a StringBuilder object, and give this function access to the StringBuilder member functions".

Good luck!

Difference between GlobalScope.launch and launch

Frevib — Sun, 04 Aug 2019 17:49:25 +0000

Both coroutine builders GlobalScope.launch {} and launch {} are used to start a new coroutine. GlobalScope.launch {} starts a new coroutine in the 'global' scope and launch {} starts a new coroutine in a CoroutineScope. But what does this mean?

To understand what is happening, we take a look at runBlocking {}. runBlocking {} will block execution (executing the next line of code after runBlocking {}) until all coroutines defined inside it's scope have been completed. For example in the following snippet, the line at 4. will only execute when both coroutines defined inside runBlocking {} have been completed:

fun main() {
    println("1. Let's begin")
    runBlocking {
        launch { 
            delay(1000)
            println("3. coroutine ONE")
        }

        launch { 
            delay(500)
            println("2. coroutine TWO")
        }
    }

    println("4. Only when the children inside runBlocking complete, execution follows on this line")
}

try it

Let try running the same code with GlobalScope.launch {} instead of launch {}:

fun main() {
    println("1. Let's start with GlobalScope.launch {}")
    runBlocking {
        GlobalScope.launch {
            delay(1000)
            println("3. coroutine ONE")
        }

        GlobalScope.launch {
            delay(100)
            println("2. coroutine TWO")
        }
    }

    println("4. This line will execute even if the coroutines inside runBlocking did not complete.")
}

try it

Now the coroutines inside runBlocking {}'s scope did not complete, and execution continued. What is happening here?

runBlocking {} defines a CoroutineScope where coroutines run in. However, the coroutines launched in the above example are running in a separate 'global' scope, where runBlocking has no control over. As far as runBlocking {} knows, there are no coroutines running inside it's scope.