From: Adam Dickmeiss <adam@indexdata.dk>
Date: Mon, 15 Feb 2010 11:53:33 +0000 (+0100)
Subject: Mutex protect the session
X-Git-Tag: v1.4.0~130
X-Git-Url: http://sru.miketaylor.org.uk/?a=commitdiff_plain;h=f6f232081df89d867621187764818ea777178e82;p=pazpar2-moved-to-github.git

Mutex protect the session

Access to the Pazpar2 session is now proteced with a Mutex. This,
for the first time, seems to produce a reasonably stable Pazpar2 in
threaded mode.
---

diff --git a/src/client.c b/src/client.c
index 3e641c5..65a308d 100644
--- a/src/client.c
+++ b/src/client.c
@@ -455,13 +455,14 @@ void client_record_response(struct client *cl)
                 else
                 {
                     struct session_database *sdb = client_get_database(cl);
+                    NMEM nmem = nmem_create();
                     const char *xmlrec;
                     char type[80];
                     if (nativesyntax_to_type(sdb, type, rec))
                         yaz_log(YLOG_WARN, "Failed to determine record type");
                     if ((xmlrec = ZOOM_record_get(rec, type, NULL)))
                     {
-                        if (ingest_record(cl, xmlrec, cl->record_offset))
+                        if (!ingest_record(cl, xmlrec, cl->record_offset, nmem))
                         {
                             session_alert_watch(cl->session, SESSION_WATCH_SHOW);
                             session_alert_watch(cl->session, SESSION_WATCH_RECORD);
@@ -471,6 +472,7 @@ void client_record_response(struct client *cl)
                     }
                     else
                         yaz_log(YLOG_WARN, "Failed to extract ZOOM record");
+                    nmem_destroy(nmem);
                 }
 
             }
diff --git a/src/http_command.c b/src/http_command.c
index 9bada03..6fe87b5 100644
--- a/src/http_command.c
+++ b/src/http_command.c
@@ -702,7 +702,7 @@ static void cmd_record(struct http_channel *c)
         return;
     }
     wrbuf_rewind(c->wrbuf);
-    if (!(rec = show_single(s->psession, idstr, &prev_r, &next_r)))
+    if (!(rec = show_single_start(s->psession, idstr, &prev_r, &next_r)))
     {
         if (session_active_clients(s->psession) == 0)
         {
@@ -732,7 +732,6 @@ static void cmd_record(struct http_channel *c)
         if (!r)
         {
             error(rs, PAZPAR2_RECORD_FAIL, "no record at offset given");
-            return;
         }
         else
         {
@@ -780,6 +779,7 @@ static void cmd_record(struct http_channel *c)
         rs->payload = nmem_strdup(c->nmem, wrbuf_cstr(c->wrbuf));
         http_send_response(c);
     }
+    show_single_stop(s->psession, rec);
 }
 
 static void cmd_record_ready(void *data)
@@ -824,7 +824,8 @@ static void show_records(struct http_channel *c, int active)
         return;
     }
 
-    rl = show(s->psession, sp, startn, &numn, &total, &total_hits, c->nmem);
+    
+    rl = show_range_start(s->psession, sp, startn, &numn, &total, &total_hits);
 
     wrbuf_rewind(c->wrbuf);
     wrbuf_puts(c->wrbuf, HTTP_COMMAND_RESPONSE_PREFIX "<show>\n<status>OK</status>\n");
@@ -856,6 +857,8 @@ static void show_records(struct http_channel *c, int active)
         wrbuf_puts(c->wrbuf, "</hit>\n");
     }
 
+    show_range_stop(s->psession, rl);
+
     wrbuf_puts(c->wrbuf, "</show>\n");
     rs->payload = nmem_strdup(c->nmem, wrbuf_cstr(c->wrbuf));
     http_send_response(c);
diff --git a/src/logic.c b/src/logic.c
index 9174fa0..a19be1f 100644
--- a/src/logic.c
+++ b/src/logic.c
@@ -98,6 +98,16 @@ static void log_xml_doc(xmlDoc *doc)
     xmlFree(result);
 }
 
+static void session_enter(struct session *s)
+{
+    yaz_mutex_enter(s->mutex);
+}
+
+static void session_leave(struct session *s)
+{
+    yaz_mutex_leave(s->mutex);
+}
+
 // Recursively traverse query structure to extract terms.
 void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
 {
@@ -125,7 +135,6 @@ void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num)
 }
 
 
-
 static void add_facet(struct session *s, const char *type, const char *value)
 {
     int i;
@@ -178,9 +187,10 @@ static xmlDoc *record_to_xml(struct session_database *sdb, const char *rec)
 
 // Add static values from session database settings if applicable
 static void insert_settings_parameters(struct session_database *sdb,
-                                       struct session *se, char **parms)
+                                       struct conf_service *service,
+                                       char **parms,
+                                       NMEM nmem)
 {
-    struct conf_service *service = se->service;
     int i;
     int nparms = 0;
     int offset = 0;
@@ -198,7 +208,7 @@ static void insert_settings_parameters(struct session_database *sdb,
             {
                 char *buf;
                 int len = strlen(val);
-                buf = nmem_malloc(se->nmem, len + 3);
+                buf = nmem_malloc(nmem, len + 3);
                 buf[0] = '\'';
                 strcpy(buf + 1, val);
                 buf[len+1] = '\'';
@@ -239,8 +249,8 @@ static void insert_settings_values(struct session_database *sdb, xmlDoc *doc,
 }
 
 static xmlDoc *normalize_record(struct session_database *sdb,
-                                struct session *se,
-                                const char *rec)
+                                struct conf_service *service,
+                                const char *rec, NMEM nmem)
 {
     xmlDoc *rdoc = record_to_xml(sdb, rec);
 
@@ -248,7 +258,7 @@ static xmlDoc *normalize_record(struct session_database *sdb,
     {
         char *parms[MAX_XSLT_ARGS*2+1];
         
-        insert_settings_parameters(sdb, se, parms);
+        insert_settings_parameters(sdb, service, parms, nmem);
         
         if (normalize_record_transform(sdb->map, &rdoc, (const char **)parms))
         {
@@ -256,7 +266,7 @@ static xmlDoc *normalize_record(struct session_database *sdb,
         }
         else
         {
-            insert_settings_values(sdb, rdoc, se->service);
+            insert_settings_values(sdb, rdoc, service);
             
             if (global_parameters.dump_records)
             {
@@ -463,18 +473,21 @@ enum pazpar2_error_code search(struct session *se,
     yaz_log(YLOG_DEBUG, "Search");
 
     *addinfo = 0;
+    
+    session_enter(se);
     nmem_reset(se->nmem);
     se->relevance = 0;
     se->total_records = se->total_hits = se->total_merged = 0;
+    reclist_destroy(se->reclist);
     se->reclist = 0;
     se->num_termlists = 0;
     live_channels = select_targets(se, filter);
-    if (live_channels)
+    if (!live_channels)
     {
-        se->reclist = reclist_create(se->nmem);
-    }
-    else
+        session_leave(se);
         return PAZPAR2_NO_TARGETS;
+    }
+    se->reclist = reclist_create(se->nmem);
 
     for (cl = se->clients; cl; cl = client_next_in_session(cl))
     {
@@ -496,6 +509,7 @@ enum pazpar2_error_code search(struct session *se,
                 client_start_search(cl);
         }
     }
+    session_leave(se);
     if (no_working == 0)
     {
         if (no_failed > 0)
@@ -616,6 +630,7 @@ void destroy_session(struct session *s)
     reclist_destroy(s->reclist);
     nmem_destroy(s->nmem);
     service_destroy(s->service);
+    yaz_mutex_destroy(&s->mutex);
     wrbuf_destroy(s->wrbuf);
 }
 
@@ -645,6 +660,8 @@ struct session *new_session(NMEM nmem, struct conf_service *service)
         session->watchlist[i].fun = 0;
     }
     session->normalize_cache = normalize_cache_create();
+    session->mutex = 0;
+    yaz_mutex_create(&session->mutex);
 
     return session;
 }
@@ -655,6 +672,7 @@ struct hitsbytarget *hitsbytarget(struct session *se, int *count, NMEM nmem)
     struct client *cl;
     size_t sz = 0;
 
+    session_enter(se);
     for (cl = se->clients; cl; cl = client_next_in_session(cl))
         sz++;
 
@@ -677,17 +695,24 @@ struct hitsbytarget *hitsbytarget(struct session *se, int *count, NMEM nmem)
         res[*count].settings_xml = w;
         (*count)++;
     }
+    session_leave(se);
     return res;
 }
 
 struct termlist_score **termlist(struct session *s, const char *name, int *num)
 {
     int i;
+    struct termlist_score **tl = 0;
 
+    session_enter(s);
     for (i = 0; i < s->num_termlists; i++)
         if (!strcmp((const char *) s->termlists[i].name, name))
-            return termlist_highscore(s->termlists[i].termlist, num);
-    return 0;
+        {
+            tl = termlist_highscore(s->termlists[i].termlist, num);
+            break;
+        }
+    session_leave(s);
+    return tl;
 }
 
 #ifdef MISSING_HEADERS
@@ -703,12 +728,13 @@ void report_nmem_stats(void)
 }
 #endif
 
-struct record_cluster *show_single(struct session *s, const char *id,
-                                   struct record_cluster **prev_r,
-                                   struct record_cluster **next_r)
+struct record_cluster *show_single_start(struct session *s, const char *id,
+                                         struct record_cluster **prev_r,
+                                         struct record_cluster **next_r)
 {
     struct record_cluster *r;
 
+    session_enter(s);
     reclist_enter(s->reclist);
     *prev_r = 0;
     *next_r = 0;
@@ -722,14 +748,21 @@ struct record_cluster *show_single(struct session *s, const char *id,
         *prev_r = r;
     }
     reclist_leave(s->reclist);
+    if (!r)
+        session_leave(s);
     return r;
 }
 
-struct record_cluster **show(struct session *s, struct reclist_sortparms *sp, 
-                             int start, int *num, int *total, Odr_int *sumhits, 
-                             NMEM nmem_show)
+void show_single_stop(struct session *s, struct record_cluster *rec)
 {
-    struct record_cluster **recs = nmem_malloc(nmem_show, *num 
+    session_leave(s);
+}
+
+struct record_cluster **show_range_start(struct session *s,
+                                         struct reclist_sortparms *sp, 
+                                         int start, int *num, int *total, Odr_int *sumhits)
+{
+    struct record_cluster **recs = nmem_malloc(s->nmem, *num 
                                                * sizeof(struct record_cluster *));
     struct reclist_sortparms *spp;
     int i;
@@ -737,6 +770,7 @@ struct record_cluster **show(struct session *s, struct reclist_sortparms *sp,
     yaz_timing_t t = yaz_timing_create();
 #endif
 
+    session_enter(s);
     if (!s->relevance)
     {
         *num = 0;
@@ -788,6 +822,11 @@ struct record_cluster **show(struct session *s, struct reclist_sortparms *sp,
     return recs;
 }
 
+void show_range_stop(struct session *s, struct record_cluster **recs)
+{
+    session_leave(s);
+}
+
 void statistics(struct session *se, struct statistics *stat)
 {
     struct client *cl;
@@ -1031,28 +1070,32 @@ static int check_record_filter(xmlNode *root, struct session_database *sdb)
 }
 
 
+static int ingest_to_cluster(struct client *cl,
+                             xmlDoc *xdoc,
+                             xmlNode *root,
+                             int record_no,
+                             const char *mergekey_norm);
+
 /** \brief ingest XML record
     \param cl client holds the result set for record
     \param rec record buffer (0 terminated)
     \param record_no record position (1, 2, ..)
-    \returns resulting record or NULL on failure
+    \retval 0 OK
+    \retval -1 failure
 */
-struct record *ingest_record(struct client *cl, const char *rec,
-                             int record_no)
+int ingest_record(struct client *cl, const char *rec,
+                  int record_no, NMEM nmem)
 {
     struct session_database *sdb = client_get_database(cl);
     struct session *se = client_get_session(cl);
-    xmlDoc *xdoc = normalize_record(sdb, se, rec);
-    xmlNode *root, *n;
-    struct record *record;
-    struct record_cluster *cluster;
-    const char *mergekey_norm;
-    xmlChar *type = 0;
-    xmlChar *value = 0;
     struct conf_service *service = se->service;
+    xmlDoc *xdoc = normalize_record(sdb, service, rec, nmem);
+    xmlNode *root;
+    const char *mergekey_norm;
+    int ret;
 
     if (!xdoc)
-        return 0;
+        return -1;
 
     root = xmlDocGetRootElement(xdoc);
 
@@ -1061,30 +1104,48 @@ struct record *ingest_record(struct client *cl, const char *rec,
         yaz_log(YLOG_WARN, "Filtered out record no %d from %s", record_no,
             sdb->database->url);
         xmlFreeDoc(xdoc);
-        return 0;
+        return -1;
     }
 
-    mergekey_norm = get_mergekey(xdoc, cl, record_no, service, se->nmem);
+    mergekey_norm = get_mergekey(xdoc, cl, record_no, service, nmem);
     if (!mergekey_norm)
     {
         yaz_log(YLOG_WARN, "Got no mergekey");
         xmlFreeDoc(xdoc);
-        return 0;
+        return -1;
     }
-    record = record_create(se->nmem, 
-                           service->num_metadata, service->num_sortkeys, cl,
-                           record_no);
-
-    cluster = reclist_insert(se->reclist, 
-                             service, 
-                             record, (char *) mergekey_norm, 
-                             &se->total_merged);
+    session_enter(se);
+    ret = ingest_to_cluster(cl, xdoc, root, record_no, mergekey_norm);
+    session_leave(se);
+
+    xmlFreeDoc(xdoc);
+
+    return ret;
+}
+
+static int ingest_to_cluster(struct client *cl,
+                             xmlDoc *xdoc,
+                             xmlNode *root,
+                             int record_no,
+                             const char *mergekey_norm)
+{
+    xmlNode *n;
+    xmlChar *type = 0;
+    xmlChar *value = 0;
+    struct session_database *sdb = client_get_database(cl);
+    struct session *se = client_get_session(cl);
+    struct conf_service *service = se->service;
+    struct record *record = record_create(se->nmem, 
+                                          service->num_metadata,
+                                          service->num_sortkeys, cl,
+                                          record_no);
+    struct record_cluster *cluster = reclist_insert(se->reclist,
+                                                    service, 
+                                                    record,
+                                                    mergekey_norm,
+                                                    &se->total_merged);
     if (!cluster)
-    {
-        /* no room for record */
-        xmlFreeDoc(xdoc);
-        return 0;
-    }
+        return -1;
     if (global_parameters.dump_records)
         yaz_log(YLOG_LOG, "Cluster id %s from %s (#%d)", cluster->recid,
                 sdb->database->url, record_no);
@@ -1285,16 +1346,12 @@ struct record *ingest_record(struct client *cl, const char *rec,
     if (value)
         xmlFree(value);
 
-    xmlFreeDoc(xdoc);
-
     relevance_donerecord(se->relevance, cluster);
     se->total_records++;
 
-    return record;
+    return 0;
 }
 
-
-
 /*
  * Local variables:
  * c-basic-offset: 4
diff --git a/src/pazpar2.h b/src/pazpar2.h
index efd743f..c814be7 100644
--- a/src/pazpar2.h
+++ b/src/pazpar2.h
@@ -116,6 +116,7 @@ struct session {
     int number_of_warnings_unknown_elements;
     int number_of_warnings_unknown_metadata;
     normalize_cache_t normalize_cache;
+    YAZ_MUTEX mutex;
 };
 
 struct statistics {
@@ -150,11 +151,16 @@ void statistics(struct session *s, struct statistics *stat);
 enum pazpar2_error_code search(struct session *s, const char *query,
                                const char *startrecs, const char *maxrecs,
                                const char *filter, const char **addinfo);
-struct record_cluster **show(struct session *s, struct reclist_sortparms *sp, int start,
-        int *num, int *total, Odr_int *sumhits, NMEM nmem_show);
-struct record_cluster *show_single(struct session *s, const char *id,
-                                   struct record_cluster **prev_r,
-                                   struct record_cluster **next_r);
+struct record_cluster **show_range_start(struct session *s,
+                                         struct reclist_sortparms *sp,
+                                         int start,
+                                         int *num, int *total, Odr_int *sumhits);
+void show_range_stop(struct session *s, struct record_cluster **recs);
+
+struct record_cluster *show_single_start(struct session *s, const char *id,
+                                         struct record_cluster **prev_r,
+                                         struct record_cluster **next_r);
+void show_single_stop(struct session *s, struct record_cluster *rec);
 struct termlist_score **termlist(struct session *s, const char *name, int *num);
 int session_set_watch(struct session *s, int what, session_watchfun fun, void *data, struct http_channel *c);
 int session_active_clients(struct session *s);
@@ -163,8 +169,7 @@ const char *session_setting_oneval(struct session_database *db, int offset);
 
 int host_getaddrinfo(struct host *host, iochan_man_t iochan_man);
 
-struct record *ingest_record(struct client *cl, const char *rec,
-                             int record_no);
+int ingest_record(struct client *cl, const char *rec, int record_no, NMEM nmem);
 void session_alert_watch(struct session *s, int what);
 void pull_terms(NMEM nmem, struct ccl_rpn_node *n, char **termlist, int *num);
 
diff --git a/src/reclists.c b/src/reclists.c
index 284d19c..5efce55 100644
--- a/src/reclists.c
+++ b/src/reclists.c
@@ -255,7 +255,8 @@ struct reclist *reclist_create(NMEM nmem)
 
 void reclist_destroy(struct reclist *l)
 {
-    yaz_mutex_destroy(&l->mutex);
+    if (l)
+        yaz_mutex_destroy(&l->mutex);
 }
 
 int reclist_get_num_records(struct reclist *l)
@@ -266,10 +267,10 @@ int reclist_get_num_records(struct reclist *l)
 }
 
 // Insert a record. Return record cluster (newly formed or pre-existing)
-struct record_cluster *reclist_insert( struct reclist *l,
-                                       struct conf_service *service, 
-                                       struct record  *record,
-                                       char *merge_key, int *total)
+struct record_cluster *reclist_insert(struct reclist *l,
+                                      struct conf_service *service, 
+                                      struct record *record,
+                                      const char *merge_key, int *total)
 {
     unsigned int bucket;
     struct reclist_bucket **p;
@@ -307,10 +308,10 @@ struct record_cluster *reclist_insert( struct reclist *l,
         new->record = cluster;
         new->hnext = 0;
         cluster->records = record;
-        cluster->merge_key = merge_key;
+        cluster->merge_key = nmem_strdup(l->nmem, merge_key);
         cluster->relevance_score = 0;
         cluster->term_frequency_vec = 0;
-        cluster->recid = merge_key;
+        cluster->recid = nmem_strdup(l->nmem, merge_key);
         (*total)++;
         cluster->metadata =
             nmem_malloc(l->nmem,
diff --git a/src/reclists.h b/src/reclists.h
index b1dcae6..bc02f10 100644
--- a/src/reclists.h
+++ b/src/reclists.h
@@ -36,10 +36,10 @@ struct reclist_sortparms
 
 struct reclist *reclist_create(NMEM);
 void reclist_destroy(struct reclist *l);
-struct record_cluster *reclist_insert( struct reclist *tl,
-                                       struct conf_service *service,
-                                       struct record  *record,
-                                       char *merge_key, int *total);
+struct record_cluster *reclist_insert(struct reclist *tl,
+                                      struct conf_service *service,
+                                      struct record  *record,
+                                      const char *merge_key, int *total);
 void reclist_sort(struct reclist *l, struct reclist_sortparms *parms);
 struct record_cluster *reclist_read_record(struct reclist *l);
 void reclist_enter(struct reclist *l);