[curl] statystyki połączenia

Generowanie statystyk połączenia:

curl -o /dev/null -s -w "DNS: %{time_namelookup}s\nConnect: %{time_connect}s\nTLS: %{time_appconnect}s\nTTFB: %{time_starttransfer}s\nTotal: %{time_total}s\n" https://en.wikipedia.org/wiki/A_Million_Random_Digits_with_100,000_Normal_Deviates
DNS: 0.020653s
Connect: 0.048270s
TLS: 0.091015s
TTFB: 0.143903s
Total: 0.183130s

[python] Random

Jednolinijkowiec na wygenerowanie 20 losowych bajtów (Ubuntu 18.04).

import random as r; r.seed(1337); " ".join([f'{b:02x}' for b in r.randbytes(20)])

Wynik:

'03 0d 25 9e 7b e3 ef ec a5 17 84 88 cd b1 ba b5 ff 3c a8 5d'

[python] asyncio - Asynchronous I/O

Asynchronieczne funkcje w Pythonie nazywane są corutines (poprzedza jest słowo kluczowe async albo są udekorowane @asyncio.coroutine). Nie można ich wołać jak zwyczajnych funkcji, trzeba skorzystać ze słowa kluczowego await (podobne do yield) i można to robić tylko wewnątrz innych corutines. await przerywa działanie i oddaje sterowanie do "event loop", które zajmuje się zarządzeniem (przekazywaniem sterowania do corutines), i w której są one rejestrowane. Przydatne linki:

• https://stackabuse.com/python-async-await-tutorial/
• https://stackoverflow.com/questions/40897428/please-explain-task-was-destroyed-but-it-is-pending/40946916#40946916

Biblioteka posiada całe mnóstwo funkcji, dla mnie najważniejsze to:

• create_task() kolejkuje zadanie
• run_until_complete() uruchomienie corutine (i wszystkie inne zakolejkowane do tej pory zadania) i czeka aż się zakończy (to konkretna). Jeżeli będą jakiś inne zadania w stanie oczekiwania to run_until_complete() nie będzie na nie czekać.
• run_forever() uruchamia wszystkie zakolejkowane zadania

Przykład:

import asyncio

async def short_task():
    print('short_task before')
    await asyncio.sleep(2)
    print('short_task after')


async def print_task():
    print('print_task')


async def long_task():
    print('long_task before')
    await asyncio.sleep(5)
    print('long_task after')


async def draw_task():
    print('draw_task')


def main():
    loop = asyncio.get_event_loop()

    loop.create_task(print_task())
    loop.create_task(long_task())

    loop.run_until_complete(short_task())
    loop.run_until_complete(draw_task())

    loop.close()


if __name__ == '__main__':
    main()

Program czeka aż zakończą się dwa zadania: short_task i draw_task, wcześniej uruchamiająć long_task. Ponieważ short_task i draw_task kończą się szybciej dostajemy ostrzeżenie, o wciąż działającym long_taks.

print_task
long_task before
short_task before
short_task after
draw_task
Task was destroyed but it is pending!
task: <Task pending coro=<long_task() done, defined at /home/beru/python_asyncio/run_until.py:13> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f47d9b217d0>()]>>

Bardziej zaawansowany przykład serwera/czatu. Tutaj mamy do czynienia z trzema rodzajami zdarzeń: uruchomienie serwera (otwarcie portów i nasłuchiwanie), oczekiwanie na tekst na stdio oraz oczekiwanie na nadejście wiadomości od klienta. Samo oczekiwanie na tekst składa się z dwóch zdarzeń: pojawienie się tekstu na stdio i zapisanie go do kolejki, oraz odczytanie, gdy coś w tej kolejce się znajduje.

Jest tu kilka kwiatków asyncio, jak np. sposób przekazywanie parametrów do obiektu Chat (przez lambdę). Ale najbardziej dokuczliwym (i dalej nie mam pewności czy zrobiłem to poprawnie) jest sposób zatrzymania programu. Po otrzymaniu komendy "exit" Chat ustawia future na True, co z kolei anuluje najpierw wszystkie zadania, następnie stopuje pętle, a na końcu pętla ta jest jeszcze zatrzymywana.
Jeżeli program jest zabijany przez Ctrl+C, to ustawienie future w obsłudze wyjątku nie zadziałała. Koniecznym stało się wywołanie explicit cancel_all_task(). Samo anulowanie zadań jest też zdaje się być asynchroniczne, więc stop() nie może być zawołane za wcześnie.

import sys
import asyncio


def main():
    queue = asyncio.Queue()
    loop = asyncio.get_event_loop()

    # Start monitoring the fd file descriptor for read availability and invoke
    # callback with the specified arguments once fd is available for reading
    loop.add_reader(sys.stdin, got_stdin_data, queue)

    fut = loop.create_future()
    fut.add_done_callback(cancel_all_task)

    coro = loop.create_server(lambda: Chat(loop, queue, fut), '127.0.0.1', 7777)
    server = loop.run_until_complete(coro)

    # Run until Ctrl+C is pressed or loop is stopped
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print('[+] Keyboard exception')
        cancel_all_task()

    # Stop server. Closing listening sockets it's done asynchronously, so
    # wait_closed() need to be used to ensure.
    server.close()
    loop.run_until_complete(server.wait_closed())

    loop.close()


def got_stdin_data(queue):
    loop = asyncio.get_event_loop()
    loop.create_task(queue.put(sys.stdin.readline()))


def cancel_all_task(result=None):
    print('[+] Cancel all tasks')
    loop = asyncio.get_event_loop()
    for task in asyncio.Task.all_tasks():
        task.cancel()
    loop.create_task(stop_loop())


async def stop_loop():
    print('[+] Stop loop')
    loop = asyncio.get_event_loop()
    loop.stop()


class Chat(asyncio.Protocol):
    def __init__(self, loop, queue, fut):
        self.loop = loop
        self.queue = queue
        self.fut = fut

    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('[+] Connection from:', peername)
        self.transport = transport
        self.loop.create_task(self._wait_for_stdin_data())

    def connection_lost(self, exc):
        print('[+] Connection lost')
        self.fut.set_result(True)

    def data_received(self, data):
        message = data.decode()
        print('[+] Data received: {!r}'.format(message))

        if message.strip() == "exit":
            self.fut.set_result(True)

    def _send_reply(self, reply):
        print('[+] Data send: {!r}'.format(reply))
        self.transport.write(reply.encode())
        self.loop.create_task(self._wait_for_stdin_data())

    async def _wait_for_stdin_data(self):
        reply = await self.queue.get()
        self._send_reply(reply)


if __name__ == '__main__':
    main()

W celu połączenia się z serwerm:

nc 127.0.0.1 7777

Działanie:

[+] Connection from: ('127.0.0.1', 45260)
[+] Data received: 'asdf\n'
[+] Data received: 'exit\n'
[+] Cancel all tasks
[+] Stop loop

OpenCV - budowanie ze źródeł

Ostatnio musiałem skomplikować OpenCV w wersji Debug. Zresztą wersja dostępna w repozytoriach Ubuntu (19.10) to obecnie 3.2, trochę stara, w porównaniu najnowszą 4.2. Pomocny link:

• https://docs.opencv.org/master/d7/d9f/tutorial_linux_install.html

A tu moje (skondensowane) kroki, żebym nie zapomniał:

# Katalog roboczy
mkdir ~/opencv_workspace
cd ~/opencv_workspace
git clone https://github.com/opencv/opencv.git
git clone https://github.com/opencv/opencv_contrib.git

# Instalacja virualenv dla Python-a. Przyda się numpy i scipy
virtualenv -p python3 venv
source venv/bin/activate
pip install numpy
pip install scipy

# Konfiguracja za pomocą CMake. 
# Wszystko co potrzebne do budowania znajdzie się w katalogu build, 
# a zainstalowane zostanie do katalogu $VIRTUAL_ENV/local/
cd opencv
mkdir build
cmake -B build/ -D CMAKE_BUILD_TYPE=Debug \
    -D OPENCV_EXTRA_MODULES_PATH=../opencv_contrib/modules/ \
    -D CMAKE_INSTALL_PREFIX=$VIRTUAL_ENV/local/ \
    -D PYTHON_EXECUTABLE=$VIRTUAL_ENV/bin/python \
    -D PYTHON_PACKAGES_PATH=$VIRTUAL_ENV/lib/python3.7/site-packages \
    -D INSTALL_PYTHON_EXAMPLES=ON

# Kompilacja i instalacja (do katalogu $VIRTUAL_ENV/local/)
cd build
make -j4
make install

Przykładowy program

#include <opencv2/opencv.hpp>
#include <iostream>

int main() {
    cv::Mat grayImg = cv::imread("color.png", cv::IMREAD_GRAYSCALE);
    cv::imwrite("gray.png", grayImg);
}

Kompilacja:

cd ~/opencv_workspace
g++ -I./venv/local/include/opencv4 -L./venv/local/lib -Wl,-rpath=./venv/local/lib \
    main.cpp \
    -lopencv_core \
    -lopencv_imgcodecs \
    -lopencv_imgproc

Wynik:

[C++17] Parallel algorithms

Biblioteka standardowa w C++17 została rozbudowana o algorytmy, których praca może zostać zrównoleglona. O sposobie wykonania decyduje ExecutionPolicy przekazane do funkcji jako argument. Programista musi zapewnić, że funkcja przekazana do "algorytmu", będzie bezpieczna - nie będzie zależności między danymi (np. modyfikowanie danych, które mogą być odczytywane przez inny równoległy wątek). Dobre wytłumaczenie różnic na stackoverflow, a tutaj małe zestawienie:

• std::execution::seq - standardowe wykonanie sekwencyjne, bez zrównoleglenia.
• std::execution::par - równoległe wykonanie (choć nie ma obietnicy, że tak się stanie).
• std::execution::par_unseq - równoległe wykonanie (choć nie ma obietnicy, że tak się stanie). Wymaga silniejszych gwarancji na to że przeplatane wywołanie funkcji będzie bezpieczne także w obrębie jednego wątku. Nowe procesory oferują taką możliwość dzięki instrukcjom do wektoryzacji - SIMD (Single-Instruction, Multiple-Data) parallelism.

Przykład z funkcją std::reduce, która w działaniu przypomina std::accumulate. Zsumowanie wartości w wektorze:

#include <iostream>
#include <numeric>
#include <execution>
#include <vector>

using namespace std;

int main() {
    vector<int> vec{1, 2, 3, 4};

    int result = std::reduce(std::execution::par,
                             begin(vec),
                             end(vec));

    cout << result << endl;
}

W przypadku gcc (9.2.1 20191008) wymagane było zainstalowanie dodatkowej paczki libtbb-dev (Threading Building Blocks).

$ sudo apt-get install libtbb-dev
$ g++ -std=c++17 main.cpp -ltbb
$ ./a.out

10

[C++] Atomics

Nie miałem do tej pory wiele do czynienia z atomic-ami (inny wpis) w prawdziwym życiu i traktuje je jako niskopoziomowy mechanizm, ale inni chyba lepiej potrafią wykorzystać ich możliwości. Pozwalają na pisanie kodu "lock-free", chociaż bez głębszego zrozumienia ich natury, niekoniecznie będzie to kod szybszy od tego opartego na muteksach. Ciekawy wykład na ich temat poprowadził Fedor Pikus na CppCon 2017: C++ atomics, from basic to advanced. What do they really do? Warto obejrzeć więcej niż raz.

Operacje na atomic-ach odzwierciedlają operacje sprzętowe i gwarantują, że zostaną wykonane w jednej transakcji (atomowo). CPU oferuje sporą liczbę mechanizmów, które są z nimi związane, z tego też względu standardowa biblioteka jest całkiem rozbudowana. Atomic-iem, może być każdy prymitywny typ (tylko takie obiekty mogą pojawić się w rejestrach CPU).
Przykłady:

// Dla
std::atomic<int> x{0};

// Operacje:
++x;           // atomowy pre-increment
x++;           // atomowy post-increment
x += 1;        // atomowy increment
int y = x * 2; // atomowy odczyt x
x = y + 2;     // atomowy zapis do x

// Uwaga, ta operacja jest niewspierana 
x *= 2;        // ERROR

// Atomowy odczyt x, po którym następuje atomowy zapis do x (dwie operacje)
x = x * 2;

W przykładzie poniżej, atomic posłużył do blokowania wątków, tak aby funkcje even/odd drukowały naprzemiennie tekst w momencie inkrementacji. Uwaga, nie ma gwarancji, że wartość counter wyświetlana na ekranie będzie zgodna z tym co było sprawdzane w if. Są to dwie atomowe operacje odczytu z pamięci, a wartość counter może się zmienić pomiędzy nimi.

#include <iostream>
#include <thread>
#include <atomic>

using namespace std;

std::atomic<int> counter{0};

void odd(size_t n) {
    for (size_t i = 0; i < n; i++) {
        if (counter % 2 == 1) {
            cout << "Odd  increment: " << counter << endl;
            counter++;
        } else {
            cout << "Odd  check: " << counter << endl;   // wartość mogła się zmienić
        }

        std::this_thread::sleep_for(std::chrono::milliseconds{20});
    }
}

void even(size_t n) {
    for (size_t i = 0; i < n; i++) {
        if (counter % 2 == 0) {
            cout << "Even increment: " << counter << endl;
            counter++;
        } else {
            cout << "Even check: " << counter << endl;   // wartość mogła się zmienić
        }
        std::this_thread::sleep_for(std::chrono::milliseconds{40});
    }
}

int main() {
    constexpr size_t steps{6};
    std::thread t1{odd, steps};
    std::thread t2{even, steps};

    t1.join();
    t2.join();
}

Wynik:

Odd  check: 0
Even increment: 0
Odd  increment: 1
Even increment: 2
Odd  increment: 3
Odd  check: 4
Even increment: 4
Odd  increment: 5
Odd  check: 6
Even increment: 6
Even check: 7
Even check: 7

[C++17] std::variant i std::visit

Nowa funkcja std::visit w C++17 umożliwia wywołanie funkcji na rzecz jakiegoś obiektu. Mając wiele takich obiektów (np. "odwiedzając" w pętli obiekty zachowane w jakimś kontenerze) mamy gotowy wzorzec wizytator (odwiedzający). Typowym sposobem implementacji takiego wzorca w C++ było skorzystanie z polimorfizmu. std::visit pozwala jednak na wywołanie funkcji na dowolnym obiekcie. Klasy nie muszą mieć wspólnego interfejsu, ale trzeba będzie skorzystać z znanego z biblioteki boost std::variant. W ten sposób można przechowywać obiekty alternatywnych typów (type-safe union).

#include <iostream>
#include <vector>
#include <variant>

using namespace std;

struct Circle {
    void show() const { cout << "Circle" << endl; }
};

struct Rect {
    void show() const { cout << "Rect" << endl; }
};

int main() {
    using Variant = std::variant<Circle, Rect>;

    vector<Variant> vec;
    vec.push_back(Circle{});
    vec.push_back(Rect{});

    for (const auto& v : vec) {
        std::visit([](const auto& obj){ obj.show(); }, v);
    }
}

Wynik:

Circle
Rect