# Lista 10
###### tags: `SO2022`


:::spoiler

:::

## Deklaracja
:::spoiler

| Zadanie | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 |
|-----------|---|---|---|---|---|---|---|---|
|Deklaracja |:heavy_check_mark:|:heavy_check_mark:|:heavy_check_mark:|:heavy_check_mark:|:heavy_check_mark:|:heavy_check_mark:|:heavy_check_mark:|:heavy_check_mark:|

:heavy_check_mark:
:::

## Zadanie 1

* ### Czym różni się przetwarzanie równoległe (ang. parallel) od przetwarzania współbieżnego (ang. concurrent)?
* **przetwarzanie równoległe** (ang. parallel):
Wiele wątków wykonuje się w tym samym czasie
* **przetwarzania współbieżnego** (ang. concurrent)
Wiele wątków wykonuje się naraz, jednak tylko jeden ma dostęp do procesora

* ### Czym charakteryzują się procedury wielobieżne (ang. reentrant)?
Procedury te mogą zostać bezpiecznie przerwane np. przerwaniem, w tym czasie ta procedura może zostać wywołana ponownie, a po powrocie z nowego wyołania, pierwsze wywołanie może być bezpiecznie kontynuowane.
Takie procedury cechują:
* Brakiem operacji na zmiennych globalnych oraz zmiennych `static`
* Brakiem możliwości modyfikacji swojego kodu
* Brakiem wywołań procedur niewielobieżnych

* ### Podaj przykład procedury w języku C
* wielobieżnej, ale nie wielowątkowo-bezpiecznej (ang. MT-safe)

        int tmp;

        void swap(int* x, int* y)
        {
            int s = tmp; 
            tmp = *x;
            *x = *y;
            *y = tmp;     

            tmp = s;
        }

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

* niewielobieżnej, ale wielowątkowo-bezpiecznej

* ### Kiedy w jednowątkowym procesie uniksowym może wystąpić współbieżność?
Taką “współbieżność” nazywamy wielozadaniowością kooperacyjną, a najlepszym przykładem są prawdopodobnie korutyny, czyli kilka współprogramów działających w obrębie jednego procesu.

## Zadanie 2

* ### Precyzyjnie opisz zjawisko:
* zakleszczenia (ang. deadlock)
to sytuacja, w której wątek A czeka na zakończenie operacji wątka B, a wątek B czeka na zakończenie wątka A. Program nigdy nie zakończy działania, gdyż wątki czekają na siebie nawzajem.

* uwięzienia (ang. livelock)
to sytuacja, w której dwa wątki zatrzymują swoje działanie aby uniknąć deadlocka, aby umożliwić innym wątkom wykonanie się, jednak robią to jednocześnie, przez co nie jest możliwe wykonanie się żadnego z nich. W przeciwieństwie do deadlocka, stan wątku może ulec zmianie.

* głodzenia (ang. starvation)
to sytuacja, w której wątek nie otrzymuje dostępu do zasobu, na który oczekuje, przez co nie może rozpocząć swojego działania.

## Zadanie 3
:::spoiler

const int n = 50;
shared int tally = 0;

void total() {
for (int count = 1; count <= n; count++)
tally = tally + 1;
}

void main() { parbegin (total(), total()); }

:::
2 <-> k * 50

## Zadanie 4

* ### Podaj procedury wątków POSIX, które pełnią podobną funkcję co `fork(2)`, `exit(3)`, `waitpid(2)`,` atexit(3)` oraz `abort(3)` na procesach.


* ### Opisz semantykę podanych funkcji i powiedz gdzie się różnią od swoich odpowiedników na procesach.
* `pthread_create(3)`

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

Tworzy nowy wątek przekazuje pointer na handler wątku, atrybuty wątku, funkcję działającą w wątku, oraz jej paramtery
* `pthread_exit(3)`

        void pthread_exit(void *retval);

Kończy działanie wątku
* `pthread_join(3)`

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

Oczekujemy na zakończenie działania wątku
* `pthread_cleanup_push(3)`

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

Dodanie wątku na stos wywołań
* `pthread_cancel(3)`

        int pthread_cancel(pthread_t thread);

Zatrzymanie wątku
* `pthread_self(3)`

        pthread_t pthread_self(void);

Pobranie id wątku

* ### Porównaj zachowanie wątków złączalnych (ang. joinable) i odczepionych (ang. detached)
* Wątki złączalne (ang. joinable)
Domyślny rodzaj wątków. Nie zwalniają zasobów po zakończeniu rutyny.
Aby je zwolnić należy wykonać `pthread_join(3)` lub zostaną zwlonione przy zakończeniu programu
* Wątki odczepione (ang. detached)
Po zakończeniu rutyny natychmiastowo zwalniają zasoby.
Aby włączyć ten tryb wątku należy użyć `pthread_detach(3)`
* to wątki, które po zakończeniu rutyny natychmiastowo zwalniają

## Zadanie 5

* ### Co nieoczekiwanego może się wydarzyć w wielowątkowym procesie uniksowym gdy:
* jeden z wątków zawoła funkcję `fork(2)` lub `execve(2)` lub `exit_group(2)`
* `fork(2)`: kopiuje cały proces, jednak posiada tylko jeden wątek
zatem z uwagi, że niektóre wątki mogą być w krytycznych częściach kodu otoczonych mutexami część danych może zostać uszkodzona w procesie dziecka
* `execve(2)`: w procesie mogą zostać otwarte deskryptory używane przez inne wątki któe nie zostaną zamknięte

* proces zadeklarował procedurę obsługi sygnału `SIGINT`, sterownik terminala wysyła do procesu `SIGINT` – w kontekście którego wątek zostanie obsłużony sygnał?
Zostanie wybrany arbitralnie(losowo) jeden wątek i tylko do niego zostanie wysłąny 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?
Odbierzemy sysgnał `SIGPIPE` i wszystkie wątki zostaną zakończone

* czytamy w wielu wątkach ze pliku zwykłego korzystając z tego samego deskryptora pliku?
Może dojść do systuacji gdy dwa razy przeczytamy ten sam fragment pliku mimo innej chęci.

## Zadanie 6

* ### Opowiedz jaka motywacja stała za wprowadzeniem wywołania `poll(2)` do systemów uniksowych?
* Możliwość efektywnego odczytywania kilku deskryptorów jednocześnie

* ### Czemu lepiej konstruować oprogramowanie w oparciu o `poll`, niż o odpytywanie deskryptorów albo powiadamianie o zdarzeniach bazujące na sygnale `SIGIO`?
* SIGIO nie mówi nam który deskryptor jest gotowy do odczytu
* Odpytywanie deskryptorów marnuje cykle procesora.

* ### Na jakich plikach można oczekiwać na zdarzenia przy pomocy `poll(2)`?
- pliki zwykłe
- terminal (tutaj trzeba uważać bo jeżeli zamknięta zostanie połowa terminala to zwracana wartość zależy od implementacji)
- pipe
- sockety

* ### Czemu wszystkie deskryptory przekazywane do `poll` powinno skonfigurować się do pracy w trybie nieblokującym?
- 
kocham ten przedmiot
- Jak poll wróci to mamy pewność, że fd X może być przeczytany raz bez blokowania. To znaczy, że na każdy read przypada 2 polle. Jeśli mamy nieblokujące fd to czytamy ile wlezie i ignorujemy EWOULDBLOCK na końcowym readzie. Jest jeszcze problem taki, że poll wróci, bo mamy klienta do zaakceptowania. Wtedy klient się rozłącza, a my blokujemy się w accept()
PS to jest od tamtych

* ### Jak zapewnić, żeby wywołania `connect(2)`, `accept(2)`, `read(2)` i `write(2)` na gnieździe sieciowym zawsze zwróciły `EWOULDBLOCK` zamiast blokować się w jądrze?
- read, write i accept: wystarczy ustawić flagę non-block na deskryptorze socketa
- Connect według manuala nie może zwrócić EWOULDBLOCK (zamiast tego jeżeli socket jest nieblokujący zwraca EINPROGRESS)

* ### Chcemy by po wywołaniu `poll(2)` pierwsza instancja wyżej wymienionych wywołań systemowych nie zwróciła `EWOULDBLOCK`. Jaka wartość musi być wpisana przez jądro do pola `revents` struktury `pollfd` dla danego deskryptora pliku, żeby mieć tę pewność?

- read i accept: POLLIN
- write: POLLOUT
- connect: ???
## Zadanie 7

#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 tid[nthreads];

  for (int i = 0; i < nthreads; i++)
    Pthread_create(&tid[i], NULL, thread, NULL);

  for (int i = 0; i < nthreads; i++)
    Pthread_join(tid[i], NULL);
  /* TODO: END */

  free(text);
}

## Zadanie 8

#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){
      char host[MAXLINE];
      char port[MAXLINE];
      socklen_t clientlen = sizeof(struct sockaddr);
      struct sockaddr clientaddr;
      int connfd;
      
      connfd = Accept(listenfd, &clientaddr, &clientlen);
      Getnameinfo(&clientaddr, clientlen, host, MAXLINE, port, MAXLINE, 0);
      addclient(connfd,host,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){
        if(clientread(i) == 0)
          delclient(i--);
        nready--;
      }
      i++;
    }
  }

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