Changeset 1619

Show
Ignore:
Timestamp:
04/16/08 16:51:54 (1 month ago)
Author:
till
Message:

nebula
- queue locks changed

Files:

Legend:

Unmodified
Added
Removed
Modified
Copied
Moved

  • nebula/trunk/src/classify.c

    r1618 r1619  
    6363                default: 
    6464                        if (FD_ISSET(cpipe[0], &rfds)) { 
     65 
    6566                                if (read(cpipe[0], &buf, 1) != 1) { 
    6667                                        fprintf(stderr, "Error - Unable to read from pipe: %s.\n", strerror(errno)); 
    6768                                        exit(EXIT_FAILURE); 
    6869                                } 
    69                         } 
    70                         pthread_mutex_lock(&classifyq_mutex); 
    71                         elem = queue_cuthead(classifyq); 
    72                         pthread_mutex_unlock(&classifyq_mutex); 
    73  
    74                         if (elem) { 
    75                                 // call classification function 
    76                                 if ((rv = classify((submission *) elem->data)) == -1) { 
    77                                         fprintf(stderr, "Error - Unable to classify submission.\n"); 
    78                                         free(elem); 
    79                                         return((void *) -1); 
    80                                 } else if (rv == 0) { 
    81                                         // known attack, do nothing 
    82                                         free(elem); 
    83                                         return((void *) 0); 
    84                                 } else { 
    85                                         // new attack 
    86                                         free(elem->data); 
    87                                         free(elem); 
     70 
     71                                pthread_mutex_lock(&classifyq_mutex); 
     72                                elem = queue_cuthead(classifyq); 
     73                                pthread_mutex_unlock(&classifyq_mutex); 
     74 
     75                                while (elem) { 
     76                                        // call classification function 
     77                                        if ((rv = classify((submission *) elem->data)) == -1) { 
     78                                                fprintf(stderr, "Error - Unable to classify submission.\n"); 
     79                                                free(elem); 
     80                                                return((void *) -1); 
     81                                        } else if (rv == 0) { 
     82                                                // known attack, do nothing 
     83                                                free(elem); 
     84                                                return((void *) 0); 
     85                                        } else { 
     86                                                // new attack 
     87                                                free(elem->data); 
     88                                                free(elem); 
     89                                        } 
     90 
     91                                        pthread_mutex_lock(&classifyq_mutex); 
     92                                        elem = queue_cuthead(classifyq); 
     93                                        pthread_mutex_unlock(&classifyq_mutex); 
    8894                                } 
    8995                        } 
    …… 
    179185                } 
    180186        } 
     187        Q_ULOCK(&clusterq->lock); 
    181188 
    182189        // connect all outliers within range 
    …… 
    184191        for (cur_hqelem = outlierq->head; cur_hqelem; cur_hqelem = cur_hqelem->next) { 
    185192                if ((score = spamsum_match(((hash*)t->data)->spamsum, ((hash*)cur_hqelem->data)->spamsum)) >= sim_threshold) { 
    186                 // unlink match from outlier list 
     193                       // unlink match from outlier list 
    187194 
    188195                        tmp_hqelem = cur_hqelem; 
    …… 
    192199                                exit(EXIT_FAILURE); 
    193200                        } 
     201 
     202 
     203                        Q_WLOCK(&clusterq->lock); 
    194204                        if (((hash*)t->data)->cl) { 
    195205                                // add other outliers to cluster 
    196                                 Q_ULOCK(&clusterq->lock); 
    197                                 Q_WLOCK(&clusterq->lock); 
    198206                                if ((qtail = add_entry_to_cluster(((hash*)t->data)->cl, tmp_hash, outlierq_max)) != NULL) { 
    199207                                        tmp_hash = qtail->data; 
    …… 
    213221                        } else { 
    214222                                // create new cluster of two outliers 
    215                                 Q_ULOCK(&clusterq->lock); 
    216                                 Q_WLOCK(&clusterq->lock); 
    217223                                if ((tmp_cqelem = queue_ins(clusterq, create_cluster(((hash*)t->data), tmp_hash, clusterhashq_max), 
    218224                                        clusterq_max)) != NULL) { 
    …… 
    243249                        } 
    244250                        Q_ULOCK(&clusterq->lock); 
    245                         Q_RLOCK(&clusterq->lock); 
    246251                } else if (score > max_score) max_score = score; 
    247252                if (!cur_hqelem) break; 
    248253        } 
    249         Q_ULOCK(&clusterq->lock); 
     254        Q_ULOCK(&outlierq->lock); 
    250255 
    251256                 

  • nebula/trunk/src/nebula.c

    r1618 r1619  
    7979int main(int argc, char *argv[]) { 
    8080        int             i, qsize, port, listen_fd, rv, gthread_no;  
    81         u_char          daemonize; 
     81        u_char          daemonize, bell; 
    8282        char            option; 
    8383        pthread_t       ntid; 
    …… 
    104104 
    105105        lock_mutex              = 1;    // if this is set, threads will lock mutexes 
     106 
     107        bell                    = 1;    // wakeup bell 
    106108 
    107109 
    …… 
    276278                        exit(EXIT_FAILURE); 
    277279                } 
    278                 if (pthread_create(&ntid, &ptattr, pt_pollgq, (void *) gthread_no)) { 
     280                if (pthread_create(&ntid, &ptattr, pt_pollgq, (void *) &gthread_no)) { 
    279281                        fprintf(stderr, "Error - Cannot create signature generation thread: %s.\n", strerror(errno)); 
    280282                        exit(EXIT_FAILURE); 
    …… 
    410412                                                 
    411413                                                // add session to classify queue 
    412                                                 // and write a byte to the itc thread to wakeup the classification thread 
    413414                                                pthread_mutex_lock(&classifyq_mutex); 
    414415                                                if (queue_append(classifyq, (void *) tmp_submission) == NULL) { 
    …… 
    416417                                                        exit(EXIT_FAILURE); 
    417418                                                }  
    418                                                 if (write(cpipe[1], "x", sizeof(u_char)) == -1) { 
     419                                                pthread_mutex_unlock(&classifyq_mutex); 
     420 
     421                                                // now send a 'wakeup byte' to the classification thread 
     422                                                if (write(cpipe[1], &bell, sizeof(u_char)) == -1) { 
    419423                                                        fprintf(stderr, "Error - Unable to write to pipe: %s.\n", strerror(errno)); 
    420424                                                        exit(EXIT_FAILURE); 
    421425                                                } 
    422                                                 pthread_mutex_unlock(&classifyq_mutex); 
    423426 
    424427                                                session_reset(&s[i], i);