# Ćwiczenia 10, grupa cz. 10-12, 5 stycznia 2023

###### tags: `SO22` `ćwiczenia` `pwit`

## Deklaracje

Gotowość rozwiązania zadania należy wyrazić poprzez postawienie X w odpowiedniej kolumnie! Jeśli pożądasz zreferować dane zadanie (co najwyżej jedno!) w trakcie dyskusji oznacz je znakiem ==X== na żółtym tle.

**UWAGA: Tabelkę wolno edytować tylko wtedy, gdy jest na zielonym tle!**

:::warning
**Uwaga**: Dzisiaj deklarujemy zadania **5 -- 9** z **listy 9** oraz **1 -- 4** z **listy 10**.

Formularz deklaracji **listy 9**:
[https://hackmd.io/@iiuwr-kba/HJ--LUuuo/edit](https://hackmd.io/@iiuwr-kba/HJ--LUuuo/edit)

:::

:::danger
| | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 |
| ----------------------:| ----- | --- | --- | --- | --- | --- | --- | --- |
Miriam Bouhajeb | | | | | | | | |
Kacper Chmielewski | | | | | X | X | X | |
Jan Jankowicz | X | X | X | | | | X | X |
Jakub Kaczmarek | | | | | | | | |
Jarosław Kadecki | X | X | X | | X | | X | |
Zuzanna Kania | X | X | X | X | X | | | |
Julia Konefał | X | X | X | | | | | |
Daniel Sarniak | | | | | | | | |
Paweł Tkocz | X | X | X | | | | X | X |
Miłosz Urbanik | X | X | X | | X | X | X | X |
Tomasz Wołczański | X | X | X | X | X | | X | |
Radosław Śliwiński | X | X | X |==X==| X | | | |
:::

:::info
**Uwaga:** Po rozwiązaniu zadania należy zmienić kolor nagłówka na zielony.
:::

## Zadanie 1
:::success
Autor: Zuzanna Kania


**Przetwarzanie współbieżne** - fragmenty wykonania wątków przeplatają się w czasie (wykonania wątków nie są sekwencyjne), możliwe nawet na procesorze jednordzeniowym

**Przetwarzanie równoległe** - fragmenty kilku wątków wykonują się dokładnie w tym samym momencie, możliwe tylko na procesorze wielordzeniowym.


---

Procedura jest **wielobieżna** (*reentrant*), jeśli może zostać przerwana w trakcie wykonywania (np. skokiem lub otrzymaniem sygnału), a następnie bezpiecznie zawołana ponownie zanim poprzednie wywołanie zdąży się zakończyć.
Procedura wielobieżna nie może:
* korzystać ze zmiennych statycznych i globalnych bez synchronizacji
* modyfikować swojego kodu bez synchronizacji
* wołać programów ani procedur, które nie są wielobieżne

Procedura jest **wielowątkowo-bezpieczna** (*thread-safe*), jeśli gwarantuje, że zawsze wykona się w sposób poprawny, niezależnie od tego, ile wątków na raz działa w programie.

Przykład procedury wielobieżnej, ale nie wielowątkowo-bezpiecznej:

int t;

void swap(int *x, int *y)
{
    int s;
    s = t;
    t = *x;
    *x = *y;
    /* Hardware interrupt might invoke isr() here. */
    *y = t;
    t = s;
}

void isr()
{
    int x = 1, y = 2;
    swap(&x, &y);
}

Przykład procedury wielowątkowo-bezpiecznej, ale nie wielobieżnej:

__thread int t;

void swap(int *x, int *y)
{
    t = *x;
    *x = *y;
    /* Hardware interrupt might invoke isr() here. */
    *y = t;
}

void isr()
{
    int x = 1, y = 2;
    swap(&x, &y);
}

---

W jednowątkowym procesie uniksowym może wystąpić współbieżność w momencie wystąpienia skoków do innych procedur (np. do signal handlera w celu obsługi sygnału).

:::

## Zadanie 2
:::success
Autor: Julia Konefał


Zakleszczenie (deadlock) : każdy wątek czeka, aż inny wątek zwolni blokadę.

Uwięzienie (livelock) : dwa lub więcej procesów nieustannie powtarza tę samą interakcję w odpowiedzi na zmiany w innych procesach, nie wykonując żadnej użytecznej pracy. Te procesy nie są w stanie oczekiwania i działają jednocześnie. Różni się to od zakleszczenia, ponieważ w zakleszczeniu wszystkie procesy są w stanie oczekiwania.

Głodzienie (starvation) : mimo dostępności zasobu, odmawiamy dostępu jakiemuś wątkowi z niskim priorytetem. Na przykład, jeśli system wielozadaniowy zawsze przełącza się między pierwszymi dwoma zadaniami, podczas gdy trzecie nigdy się nie uruchamia, to trzecie zadanie jest głodzone (pozbawione czasu procesora).


:::

## Zadanie 3
:::success
Autor: Jarosław Kadecki

:::
Po deassemblacji fragment kodu odpowiedzialny za inkrementację tally wyglada następująco:
Możemy zauważyć, że najpierw wczytujemy wartość tally z jego lokalizacji w pamięci, natępnie dodajemy 1 i ponownie umieszczamy wartość pod adresem tally. Oznacza to, że może zaistnieć sytuacja kiedy następująco:
1. Proc1 odczytuje wartość tally=0;
2. Proc2 n-1 razy odczytuje wartość tally, zwiększa ją o 1 i wpisuje wynik pod adres tej zmiennej.
Wtedy tally = n-1
3. Proc1 zwiększa wartość którą odczytał o 1 (czyli 0) i wpisuje wynik pod adres zmiennej.
Wtedy tally = 1.
4. Proc2 (wykonując n-tą iteracje pętli) odczytuje tally.
5. Proc1 wykonuje n-1 razy odczytuje, inkrementuje i zapisuje tally.
6. Proc2 wpisuje i zwięksa wartość tally którą odczytał (1) o 1 i dokonuje zapisu.

Wtedy wartość tally zwiększa się tylko o 2. Natomiast jeśli działanie procesów odbyło się jeden po drugim to ostatecznie wartość tally byłaby równa 2n.

Jeśli wystartujemy K procesów może nastąpić podobny przeplot. Tak samo wykonuje się krok 1 i 2. Następnie zwoją pracę wykonują Proc3 - ProcK. Później mają miejsce kroki 3-6. Wtedy końcowa wartość tally = 2.

Jeśli działanie procesów odbyłoby się jeden po drugim to ostatecznie wartość tally byłaby równa Kn.

## Zadanie 4
:::success
Autor: Radosław Śliwiński
Odpowiedniki:
* `fork()` -> `pthread_create()`
* `exit()` -> `pthread_exit()`
* `waitpid()` -> `pthread_join()`
* `atexit()` -> `pthread_cleanup_push()`
* `abort()` -> `pthread_cancel()`

int pthread_create(pthread_t *restrict thread,
                          const pthread_attr_t *restrict attr,
                          void *(*start_routine)(void *),
                          void *restrict arg);

noreturn void pthread_exit(void *retval);

int pthread_join(pthread_t thread, void **retval);

void pthread_cleanup_push(void (*routine)(void *), void *arg);

int pthread_cancel(pthread_t thread);

Wątek jest `złączalny` kiedy inny wątek może uzyskać jego return status przy użyciu `pthread_join()` i jest to domyślne działanie wątków. Czasami nie jest potrzebny return status danego procesu. W tym przypadku proces możemy oznaczyć jako `odczepiony`. Z `odczepionego` procesu nie da się uzyskać jego return status i nie da się z powrotem go zmienić na `złączalny`. Poza tym główną różnicą pomiędzy tymi dwoma stanami jest czas kiedy wątki zwalniają zasoby ze stosu. W wątku `odczepionym` są one zwalniane automatycznie przez system przy terminacji wątku, a w przypadku wątków `złączalnych` są one zwalniane gdy program kończy działanie. Odpowiedzialność za zwolnienie zasobów tego wątku spoczywa na programiście jeżeli chce użyć miejsca w trakcie wykonania programu.
:::

## Zadanie 5
:::success
Autor: Kacper Chmielewski

**Jeden z wątków zawoła funkcje:**
- fork - kopiuje cały proces, jednak posiada tylko jeden wątek zatem z uwagi, że niektóre wątki mogą być mieć założone takie same blokady w obrębie jądra przez co nie będziemy pewni co dokładnie ma się wykonać
- execve - w procesie mogą zostać otwarte deskryptory używane przez inne wątki które nie zostaną zamknięte
- exit_group - zostaną zamknięte wszystkie wątki

**Proces zadeklarował procedurę obsługi sygnały ```SIGINT```, sterownik terminala wysyła do procesu ```SIGINT``` - w kontekście którego wątek zostanie obsłużony sygnał?**

Zostanie losowo wybrany jeden wątek i tylko on otrzyma sygnał.

**Określono domyślną dyspozycję sygnału ```SIGPIPE```, a jeden z wątków spróbuje zapisać do rury pipe(2), której drugi koniec został zamknięty?**

Sygnał zostanie odebrany, a wątki zostaną zakończone

**Czytamy w wielu wątkach ze pliku zwykłego korzystając z tego samego deskryptora pliku?**
Kilka wątków będzie mogło przeczytać ten sam fragment pliku.
:::

## Zadanie 6
:::success
Autor: Miłosz Urbanik

:::
Multiplexing I/O pozwala na wygodne i wydajne oczekiwanie na dostępność odpowiednich operacji na wielu deskryptorach plików.

Do obsługi wielu deskryptorów plików, odpytując je w bezpośrednio w trybie blokującym, należałoby używać tyle wątków/procesów co liczba deskryptorów. W trybie nieblokującym, marnujemy czas procesora i musimy "zgadywac" ile czasu czekać do kolejnego obrotu pętli w której odpytamy deskryptory.

Powiadamianie o zdarzeniach bazujących na sygnałach ma ten minus, że sygnał nie informuje nas jakie deskryptory i w jaki sposób zmieniły swój status (pozwalają na odczyt/zapis danych) - musimy zatem przejrzeć je wszystkie. Dodatkowo obsługa sygnałów jest bardziej skomplikowana - wymaga korzystania z procedur obsługująych sygnały, a operacje na deskryptorach powinny być nieblokujące.

Poll obsługuje:
- pliki zwykłe (nie ma zbyt wiele sensu POLLIN i POLLOUT zawsze zapalone)
- terminale i pseudoterminale
- rury i rury FIFO
- gniazda
- niektóre urządzenia znakowe

Deskryptory plików powinny być przekazywane w trybie nieblokującym, żeby uniknąć blokowania podczas operacji na nich po powrocie z polla.
Aby gniazda nie blokowały, musimy ustawić ich deskryptor plików w trybie nieblokujących przy użyciu funkcji `fctl`

flags = Fcntl(fd, F_GETFL, 0);
Fcntl(fd, F_SETFL, flags | O_NONBLOCK);

Jeśli deskryptor ma w swojej masce revents zapalony bit `POLLIN`, to nie otrzymamy `errno == EWOULDBLOCK`

## Zadanie 7
:::success
Autor: Tomasz Wołczański
:::

#include "csapp.h"
#include "rio.h"

#define THREADMAX 100
#define ITERMAX 50

/* Read whole file into process address space.
 * Make sure it ends with a newline and NUL character. */
static char *readtext(const char *path) {
  int fd = Open(path, O_RDONLY, 0);

  struct stat sb;
  Fstat(fd, &sb);

  size_t len = sb.st_size;
  char *data = Malloc(len + 2);
  Read(fd, data, len);

  Close(fd);

  if (data[len - 1] != '\n')
    data[len++] = '\n';
  data[len] = '\0';

  return data;
}

typedef struct line {
  const char *data;
  size_t length;
} line_t;

static const size_t splitlines(const char *text, line_t **linesp) {
  size_t capacity = 10;
  size_t nlines;

  line_t *lines = Malloc(capacity * sizeof(line_t));
  
  for (nlines = 0; *text; nlines++) {
    /* Grow lines array if needed. */
    if (nlines == capacity) {
      capacity *= 2;
      lines = Realloc(lines, capacity * sizeof(line_t));
    }

    const char *begin = text;

    /* Advance to the next line. */
    while (*text && *text != '\n')
      text++;

    /* Skip newline character. */
    if (*text == '\n')
      text++;

    lines[nlines].data = begin;
    lines[nlines].length = text - begin;
  }

  /* Trim lines array down. */
  *linesp = Realloc(lines, nlines * sizeof(line_t));
  return nlines;
}

static sig_atomic_t quit = false;

static void sigint_handler(int sig) {
  safe_printf("Quit requested!\n");
  quit = true;
}

static char *c_host = NULL;
static char *c_port = NULL;
static line_t *c_lines = NULL;
size_t c_nlines = 0;

static void *thread(void *data) {
  unsigned seed = (unsigned)pthread_self();
  char *buf = NULL;
  size_t buflen = 0;

  while (!quit) {
    int fd = Open_clientfd(c_host, c_port);

    for (int i = 0; i < ITERMAX && !quit; i++) {
      line_t *line = &c_lines[rand_r(&seed) % c_nlines];

      if (buflen < line->length) {
        buflen = line->length;
        buf = Realloc(buf, buflen);
      }

      Rio_writen(fd, (void *)line->data, line->length);
      Rio_readn(fd, buf, line->length);
      assert(memcmp(buf, line->data, line->length) == 0);
      usleep(1000 * (rand_r(&seed) % 50));
    }

    Close(fd);
  }

  free(buf);
  return NULL;
}

int main(int argc, char **argv) {
  if (argc != 5)
    app_error("usage: %s <textfile> <nthreads> <host> <port>\n", argv[0]);

  size_t nthreads = atol(argv[2]);
  assert(nthreads > 0 && nthreads <= THREADMAX);

  char *text = readtext(argv[1]);

  c_nlines = splitlines(text, &c_lines);
  c_host = argv[3];
  c_port = argv[4];

  Signal(SIGINT, sigint_handler);

  /* TODO: Start threads and wait for them to finish. */
  pthread_t *threads = Malloc(nthreads * sizeof(pthread_t));
  for (int i = 0; i < nthreads; i++)
    Pthread_create(&threads[i], NULL, thread, NULL);
  for (int i = 0; i < nthreads; i++)
    Pthread_join(threads[i], NULL);

  free(c_lines);
  free(text);
}

## Zadanie 8
:::success
Autor: Paweł Tkocz
:::

#include "csapp.h"
#include "rio.h"

#define LISTENQ 10

typedef struct client {
  size_t id;
  rio_t rio;
} client_t;

static size_t clientid = 1;
static client_t **clients;
static struct pollfd *fds;
static nfds_t nfds = 0;
static nfds_t maxfds = 0;

static sig_atomic_t quit = false;
static size_t nbytes = 0; /* Counts total bytes received by server */

static void initclients(int listenfd) {
  nfds = 1;
  maxfds = 1;
  fds = Malloc(sizeof(struct pollfd));
  fds[0].fd = listenfd;
  fds[0].events = POLLIN;
}

static void addclient(int connfd, const char *hostname, const char *port) {
  printf("[%ld] Connected to %s:%s\n", clientid, hostname, port);

  /* Should expand space to accommodate for new client ? */
  if (maxfds == nfds) {
    maxfds++;
    fds = Realloc(fds, sizeof(struct pollfd) * maxfds);
    clients = Realloc(clients, sizeof(client_t *) * maxfds);
    clients[maxfds - 1] = Malloc(sizeof(client_t));
  }

  int i = nfds++;
  fds[i].fd = connfd;
  fds[i].events = POLLIN; /* Wait only for input events! */
  fds[i].revents = 0;

  client_t *c = clients[i];
  c->id = clientid++;
  rio_readinitb(&c->rio, connfd);
}

static void delclient(int i) {
  assert(i > 0);
  Close(fds[i].fd);
  printf("[%ld] Disconnected!\n", clients[i]->id);

  if (i < nfds - 1) {
    fds[i] = fds[nfds - 1];

    /* Swap slots so that prefix of array comprises only active clients. */
    client_t *tmp = clients[i];
    clients[i] = clients[nfds - 1];
    clients[nfds - 1] = tmp;
  }

  nfds--;
}

static int clientread(int i) {
  char buf[MAXLINE];
  int n = Rio_readlineb(&clients[i]->rio, buf, MAXLINE);

  if (n > 0) {
    Rio_writen(fds[i].fd, buf, n);
    nbytes += n;
    printf("[%ld] Received %d (%ld total) from client\n",
           clients[i]->id, n, nbytes);
  }

  return n;
}

static void sigint_handler(int sig) {
  safe_printf("Server received quit request!\n");
  quit = true;
}

int main(int argc, char **argv) {
  if (argc != 2)
    app_error("usage: %s <port>\n", argv[0]);

  Signal(SIGINT, sigint_handler);

  int listenfd = Open_listenfd(argv[1], LISTENQ);
  initclients(listenfd);

  while (!quit) {
    int nready = Poll(fds, nfds, 500);
    if (nready == 0)
      continue;

    /* TODO: If listening descriptor ready, add new client to the pool. */
    if (fds[0].revents & POLLIN){
      struct sockaddr_storage clientaddr;
      socklen_t clientlen = sizeof(struct sockaddr_storage);
      int connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);

      char client_hostname[MAXLINE], client_port[MAXLINE];
      Getnameinfo((SA *)&clientaddr, clientlen, client_hostname, MAXLINE, client_port, MAXLINE, 0);
      addclient(connfd, client_hostname, client_port);

      nready--;
    }
    /* TODO: Echo a text line from each ready connected descriptor.
     * Delete a client when end-of-file condition was detected on socket. */
    int i = 1;
    while (nready > 0) {
      if (fds[i].revents & POLLIN)
      {
        int bytes_read = clientread(i);
        if (bytes_read == 0)
        {
          delclient(i);
          i--;
        }
        nready--;
      }
      i++;
    }
  }

  printf("Server received %ld total bytes.\n", nbytes);
  return EXIT_SUCCESS;
}