From f6f232081df89d867621187764818ea777178e82 Mon Sep 17 00:00:00 2001 From: Adam Dickmeiss Date: Mon, 15 Feb 2010 12:53:33 +0100 Subject: [PATCH] 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. --- src/client.c | 4 +- src/http_command.c | 9 ++- src/logic.c | 163 +++++++++++++++++++++++++++++++++++----------------- src/pazpar2.h | 19 +++--- src/reclists.c | 15 ++--- src/reclists.h | 8 +-- 6 files changed, 143 insertions(+), 75 deletions(-) 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 "\nOK\n"); @@ -856,6 +857,8 @@ static void show_records(struct http_channel *c, int active) wrbuf_puts(c->wrbuf, "\n"); } + show_range_stop(s->psession, rl); + wrbuf_puts(c->wrbuf, "\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); -- 1.7.10.4