Newsgroup: it.comp.lang.c
Titolo: Cerco conferme (thread+unix)

Salve, sono in cerca di conferme... :-)
premessa: sto lavorando in C su un server Compaq multiprocessore (4
alpha)con sistema operativo Tru64 Unix.
Lo scenario su cui sto lavorando attualmente e' il seguente: una
procedura prende in input un file ascii (molto lungo circa 12.000.000
di righe) e passa ogni singola linea ad un 'oggetto' (NON inteso come
istanza di una classe eh :-), ma come una serie di procedure autonome
)che la tratta in maniera particolare e produce in output (in append)
un'altra riga un file ascii esterno.
Attualmente il sistema funziona perfettamente ed ha un tempo di
esecuzione <T>
Ho notato che, se lancio in parallelo da due a quattro processi con
files di input diversi e stesso file di output, il tempo di esecuzione
rimane pressappoco <T>, cioe' i processi hanno lavorato PERFETTAMENTE
in parallelo.
Ho pensato, quindi, di estendere questo parallelismo all'interno della
lavorazione del singolo file utilizzando i threads per ridurre il
tempo di elaborazione.
L'architettura che avevo pensato era la seguente, tenendo presente che
il tempo impiegato dall 'oggetto' per processare la riga e' piu' alto
di quello impiegato per la lettura della riga:
- <X> threads <master> con il compito di leggere linee dal file di
input e copiarle in un buffer (ad es: una FIFO di max <K> linee in
memoria )
(anche uno solo dovrebbe bastare, perche' questa fase e' veloce)
- <Y> threads <slave :-)> che scodano dalla FIFO e producono l'output.
Chiaramente i valori di <X>, <Y> e <Z> sono da testare empiricamente
direttamente sulla macchina  :-)
Ho appena finito di studiarmi la Posix Threads Library e ho provato a
scrivere qualche esempio utilizzando thread e mutex, ma nn sono
riuscito a produrne di accettabili... il problema credo che sia nei
meccanismi impliciti di lock/unlock su file...
Qualcuno riesce a farmi un esempio di questa architettura ipotizzando
che 'oggetto' nn faccia altro che aggiungere alla riga letta un
carattere ?
Ciao,
Stefano.

 file...
Nessun lock  implicito, se ti serve  te lo metti da  te. Nel nostro caso
non serve...
Io utilizzerei  due ring buffer  (uno di input e  uno di output)  per la
comunicazione tra i  vari thread, una variabile  globale per controllare
l'esecuzione, una funzione di estrazione temporizzata.
I ring buffer verrano gestiti  come uno stack, con inserimento bloccante
ed estrazione temporizzata.
#define BUFFER_SIZE 50
typedef struct {
        char *data[BUFFER_SIZE];
        int readpos, writepos;
        pthread_mutex_t lock;
        pthread_cond_t notempty;
        pthread_cond_t notfull;
} t_stack;
/* Inizializzo i puntatori, il lock e le variabili condizionali */
void init_stack (t_stack *stack) {
        pthread_mutex_init(&stack->lock, NULL);
        pthread_cond_init(&stack->notempty, NULL);
        pthread_cond_init(&stack->notfull, NULL);
        stack->readpos = stack->writepos = 0;
}
/* Inserimento */
void push(t_stack *stack, char *line) {
        /* Acquisisco il lock sul buffer */
        pthread_mutex_lock(&stack->lock);
        while ((stack->writepos + 1) % BUFFER_SIZE == stack->readpos)
                /* Buffer pieno. Rilascio il lock ed aspetto che venga 
                 * segnalata la condizione 'notfull'
                 */
                pthread_cond_wait(&stack->notfull, &stack->lock);
        stack->buffer[stack->writepos] = data;
        stack->writepos++;
        if (stack->writepos >= BUFFER_SIZE)
                stack->writepos = 0;
        /* Segnalo che lo stack non e` vuoto */
        pthread_cond_signal(&stack->notempty);
        /* Rilascio il lock */
        pthread_mutex_unlock(&stack->lock);
}
/* Estrazione */
char *pop(t_stack *stack) {
        char *data;
        int ret;
        struct timeval now;
        struct timespec timeout;
 
        pthread_mutex_lock(&stack->lock);
        /* Timeout di 3 secondi */
        gettimeofday(&now, NULL);
        timeout.tv_sec = now.tv_sec + 3;
        timeout.tv_nsec = now.tv_usec * 1000;
        while(stack->readpos == stack->writepos) {
                /* Aspetto i dati per tre secondi, se non ricevo nulla
                 * esco
                 */
                 ret = pthread_cond_timedwait(&stack->notempty, 
                        &stack->lock, &timeout);
                if (ret == ETIMEDOUT) {
                        /* Timeout */
                        pthread_mutex_unlock(&stack->lock);
                        return NULL;
                }
        }
        data = stack->buffer[stack->readpos];
        stack->readpos++;
        if (stack->readpos >= BUFFER_SIZE)
                stack->readpos = 0;
        /* Segnalo che il buffer non e` pieno */
        pthread_cond_signal(&stack->notfull);
        pthread_mutex_unlock(&stack->lock);
        return data;
}
void destroy_stack(t_stack *stack) {
        pthread_cond_destroy(&stack->notfull);
        pthread_cond_destroy(&stack->notempty);
        pthread_mutex_destroy(&stack->lock);
}
A questo  punto direi che  il grosso  del lavoro e`  fatto. Bisogna solo
creare i thread:
#define SLAVE 4
t_stack out;
t_stack in;
int status = 0;
pthread_mutex_t status_lock = PTHREAD_MUTEX_INITIALIZER;
void *slave(void *useless) {
        char *line;
        while(1) {
                line = pop(&in);
                if (art == NULL) {
                        int tmp;
 
                        pthread_mutex_lock(&status_lock);
                        tmp = status;
                        pthread_mutex_unlock(&status_lock);
                        if (tmp == 0) {
                                /* Devo continuare, cedo il processore */
                                sched_yield();
                                continue;
                        }
                        /* Devo uscire, ricontrollo lo stack ed esco */
                        if ((art = pop(&art_s)) == NULL)
                                pthread_exit(NULL);
                        /* Il buffer non era vuoto, continuo */
                }
 
                /* Elaborazione */
                do_something(line);
                push(&out, &line);
        }
}
void *write_thread(void *useless) {
        char *line;
        FILE *outfile;
        outfile = fopen(...);
        while(1) {
                line = pop(&out);
                if (line == NULL) {
                        int tmp;
 
                        pthread_mutex_lock(&status_lock);
                        tmp = status;
                        pthread_mutex_unlock(&status_lock);
                        if (tmp != 2) {
                                sched_yield();
                                continue;
                        }
                        /* Devo uscire */
                        if ((line = pop(&out)) == NULL) {
                                fclose(outfile);
                                pthread_exit(NULL);
                        }
                }
 
                fputs(line, outfile);
                free(line);
        }
}
 
int main() {
        FILE *infile;
        ptread_t th[SLAVE + 1];
        int i;
        char buff[MAX_LINE_SIZE];
        infile = fopen(...);
        for (i = 0; i < SLAVE; i++)
                pthread_create(&th[i], NULL, slave, NULL);
        pthread_create(&th[i], NULL, write_thread, NULL);
        while(fgets(buff, sizeof(buff), infile)) {
                char *p = malloc(strlen(buff) + 1);
                push(&in, p);
        }
        /* Fine lettura */
        pthread_lock(&status_lock);
        status++;
        pthread_unlock(&status_lock);
 
        for (i = 0; i < SLAVE; i++)
                pthread_join(th[i], NULL);
        pthread_lock(&status_lock);
        status++;
        pthread_unlock(&status_lock);
        pthread_join(th[i], NULL);
        return 0;
}
Questo dovrebbe andare  bene come base per il  tuo programma. Occhio che
e` codice non compilato, potrebbero esserci errori.
In pratica il main  crea i thread ed inizia a  leggere il file, mettendo
le righe lette  nel buffer 'in'. I thread slave estraggono  le linee dal
buffer  'in',  fanno  quello  che  devono  fare  e  rimettono  la  linea
modificata nel  buffer 'out'. Il  thread 'write_thread' estrae  le linee
dal buffer 'out', le scrive e libera la memoria occupata.
ciao
Kronos
-- 
Home: 
http://kronoz.cjb.net
It can't rain forever,
but you can die before seeing the sun again.

 Kronos
Ciao grazie ancora, 
Stefano.

 aggirare il problema ?
Devi  utilizzare  la  funzione  __fsetlocking  per  rimuovere  il  lock.
Attenzione  che il  lock viene  effettuato solo  sull'oggetto FILE,  non
sull'inode  del file. Il  lock  e` quindi  efficace  quando piu`  thread
utilizzando lo stesso oggetto FILE, non quando ognuno apre per conto suo
il  file. Inoltre,  anche  se  il  lock  garantisce  l'atomicita`  delle
operazioni  di  IO sul  file,  l'ordine  con queste  operazioni  vengono
effettuate  non e`  determinabile  a priori. Finche`  ti  limiti solo  a
scrivere (o solo a leggere)  non dovrebbero esserci problemi. Se inizi a
mischiare letture,  scritture e magari spostamenti  all'interno del file
avrai certamente dei problemi.
ciao
Kronos
-- 
Home: 
http://kronoz.cjb.net
Non capisco tutta questa eccitazione per il Multitasking: 
io sono anni che leggo in bagno.