Add support for thread config element which specifies number of
[yazproxy-moved-to-github.git] / src / yaz-proxy.cpp
index adc5208..194b973 100644 (file)
@@ -1,4 +1,4 @@
-/* $Id: yaz-proxy.cpp,v 1.56 2006-04-13 00:41:11 adam Exp $
+/* $Id: yaz-proxy.cpp,v 1.68 2006-06-09 09:35:14 adam Exp $
    Copyright (c) 1998-2006, Index Data.
 
 This file is part of the yazproxy.
@@ -59,9 +59,6 @@ using namespace yazpp_1;
 #define strncasecmp _strnicmp
 #endif
 
-#define USE_AUTH_MSG 1
-
-#if USE_AUTH_MSG
 class YAZ_EXPORT Auth_Msg : public IMsg_Thread {
 public:
     int m_ret;
@@ -106,7 +103,7 @@ IMsg_Thread *Auth_Msg::handle()
 
 void Auth_Msg::result()
 {
-    if (m_proxy->dec_ref(false))
+    if (m_proxy->dec_ref())
     {
         yaz_log(YLOG_LOG, "Auth_Msg::proxy deleted meanwhile");
     }
@@ -122,8 +119,6 @@ void Auth_Msg::result()
     delete this;
 }
 
-#endif
-
 void Yaz_Proxy::result_authentication(Z_APDU *apdu, int ret)
 {
     if (apdu == 0 || ret == 0)
@@ -131,7 +126,7 @@ void Yaz_Proxy::result_authentication(Z_APDU *apdu, int ret)
         Z_APDU *apdu_reject = zget_APDU(odr_encode(), Z_APDU_initResponse);
         *apdu_reject->u.initResponse->result = 0;
         send_to_client(apdu_reject);
-        dec_ref(false);
+        dec_ref();
     }
     else
     {
@@ -295,6 +290,7 @@ Yaz_Proxy::Yaz_Proxy(IPDU_Observable *the_PDU_Observable,
     m_ref_count = 1;
     m_main_ptr_dec = false;
     m_peername = 0;
+    m_num_msg_threads = 0;
 }
 
 void Yaz_Proxy::inc_ref()
@@ -358,7 +354,8 @@ int Yaz_Proxy::set_config(const char *config)
     {
         int period = 60;
         m_config->get_generic_info(&m_log_mask, &m_max_clients,
-                                   &m_max_connect, &m_limit_connect, &period);
+                                   &m_max_connect, &m_limit_connect, &period,
+                                   &m_num_msg_threads);
         m_connect.set_period(period);
     }
     return r;
@@ -414,7 +411,7 @@ Yaz_ProxyConfig *Yaz_Proxy::check_reconfigure()
                 int period = 60;
                 cfg->get_generic_info(&m_log_mask, &m_max_clients,
                                       &m_max_connect, &m_limit_connect,
-                                      &period);
+                                      &period, &m_num_msg_threads);
                 m_connect.set_period(period);
             }
         }
@@ -432,12 +429,16 @@ IPDU_Observer *Yaz_Proxy::sessionNotify(IPDU_Observable
 
     char session_str[200];
     const char *peername = the_PDU_Observable->getpeername();
+    if (!peername)
+        peername = "nullpeer";
+
     if (m_log_mask & PROXY_LOG_IP_CLIENT)
         sprintf(session_str, "%ld:%d %.80s %d ",
                 (long) time(0), m_session_no, peername, 0);
     else
         sprintf(session_str, "%ld:%d %d ",
                 (long) time(0), m_session_no, 0);
+    m_session_no++;
 
     yaz_log (YLOG_LOG, "%sNew session %s", session_str, peername);
 
@@ -451,9 +452,14 @@ IPDU_Observer *Yaz_Proxy::sessionNotify(IPDU_Observable
     new_proxy->set_default_target(m_default_target);
     new_proxy->m_max_clients = m_max_clients;
     new_proxy->m_log_mask = m_log_mask;
+    new_proxy->m_session_no = m_session_no;
+    new_proxy->m_num_msg_threads = m_num_msg_threads;
 
+#if 0
+    // in case we want to watch a particular client..
     if (!strcmp(peername, "tcp:163.121.19.82")) // NIS GROUP
         new_proxy->m_log_mask = 255;
+#endif
 
     new_proxy->set_APDU_log(get_APDU_log());
     if (new_proxy->m_log_mask & PROXY_LOG_APDU_CLIENT)
@@ -465,8 +471,12 @@ IPDU_Observer *Yaz_Proxy::sessionNotify(IPDU_Observable
     new_proxy->set_proxy_negotiation(m_proxy_negotiation_charset,
         m_proxy_negotiation_lang, m_proxy_negotiation_default_charset);
     // create thread object the first time we get an incoming connection
-    if (!m_my_thread)
-        m_my_thread = new Msg_Thread(m_socket_observable, 1);
+    if (!m_my_thread && m_num_msg_threads > 0)
+    {
+        yaz_log (YLOG_LOG, "%sStarting message thread management. number=%d",
+                 session_str, m_num_msg_threads);
+        m_my_thread = new Msg_Thread(m_socket_observable, m_num_msg_threads);
+    }
     new_proxy->m_my_thread = m_my_thread;
     return new_proxy;
 }
@@ -600,8 +610,14 @@ Yaz_ProxyClient *Yaz_Proxy::get_client(Z_APDU *apdu, const char *cookie,
             m_client_idletime = client_idletime;
             timeout(m_client_idletime);
         }
+
+        // get those FILE descriptors available 
+        m_parent->low_socket_close();
         if (cql2rpn_fname)
             m_cql2rpn.set_pqf_file(cql2rpn_fname);
+        // reserve them again
+        m_parent->low_socket_open();
+        
         if (negotiation_charset || negotiation_lang || default_client_query_charset)
         {
             set_proxy_negotiation(negotiation_charset,
@@ -733,7 +749,7 @@ Yaz_ProxyClient *Yaz_Proxy::get_client(Z_APDU *apdu, const char *cookie,
                 yaz_log (YLOG_LOG, "%sMAXCLIENTS %d Destroy %d",
                          m_session_str, parent->m_max_clients, c->m_seqno);
                 if (c->m_server && c->m_server != this)
-                    c->m_server->dec_ref(true);
+                    c->m_server->dec_ref();
             }
             else
             {
@@ -748,7 +764,7 @@ Yaz_ProxyClient *Yaz_Proxy::get_client(Z_APDU *apdu, const char *cookie,
                 if (c->m_server && c->m_server != this)
                 {
                     c->m_server->m_client = 0;
-                    c->m_server->dec_ref(true);
+                    c->m_server->dec_ref();
                 }
                 (parent->m_seqno)++;
                 c->m_target_idletime = m_target_idletime;
@@ -1610,8 +1626,21 @@ Z_APDU *Yaz_Proxy::result_set_optimize(Z_APDU *apdu)
         {
             Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentResponse);
             new_apdu->u.presentResponse->records =
-                create_nonSurrogateDiagnostics(odr_encode(), 30,
-                                               pr->resultSetId);
+                create_nonSurrogateDiagnostics(
+                    odr_encode(), 
+                    YAZ_BIB1_SPECIFIED_RESULT_SET_DOES_NOT_EXIST,
+                    pr->resultSetId);
+            send_to_client(new_apdu);
+            return 0;
+        }
+        if (start < 1 || toget < 0)
+        {
+            Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentResponse);
+            new_apdu->u.presentResponse->records =
+                create_nonSurrogateDiagnostics(
+                    odr_encode(), 
+                    YAZ_BIB1_PRESENT_REQUEST_OUT_OF_RANGE, 
+                    0);
             send_to_client(new_apdu);
             return 0;
         }
@@ -1621,7 +1650,10 @@ Z_APDU *Yaz_Proxy::result_set_optimize(Z_APDU *apdu)
             {
                 Z_APDU *new_apdu = create_Z_PDU(Z_APDU_presentResponse);
                 new_apdu->u.presentResponse->records =
-                    create_nonSurrogateDiagnostics(odr_encode(), 13, 0);
+                    create_nonSurrogateDiagnostics(
+                        odr_encode(), 
+                        YAZ_BIB1_PRESENT_REQUEST_OUT_OF_RANGE,
+                        0);
                 send_to_client(new_apdu);
                 return 0;
             }
@@ -1667,6 +1699,21 @@ Z_APDU *Yaz_Proxy::result_set_optimize(Z_APDU *apdu)
 
     this_query->set_Z_Query(sr->query);
 
+    // Check for non-negative piggyback params.
+    if (*sr->smallSetUpperBound < 0
+        || *sr->largeSetLowerBound < 0
+        || *sr->mediumSetPresentNumber < 0)
+    {
+        Z_APDU *new_apdu = create_Z_PDU(Z_APDU_searchResponse);
+        // Not a present request.. But can't find better diagnostic
+        new_apdu->u.searchResponse->records =
+            create_nonSurrogateDiagnostics(
+                odr_encode(), 
+                YAZ_BIB1_PRESENT_REQUEST_OUT_OF_RANGE, 0);
+        send_to_client(new_apdu);
+        return 0;
+    }
+
     char query_str[120];
     this_query->print(query_str, sizeof(query_str)-1);
     yaz_log(YLOG_LOG, "%sSearch %s", m_session_str, query_str);
@@ -1687,7 +1734,7 @@ Z_APDU *Yaz_Proxy::result_set_optimize(Z_APDU *apdu)
 
             if (toget > m_client->m_last_resultCount)
                 toget = m_client->m_last_resultCount;
-
+            
             if (sr->mediumSetElementSetNames)
             {
                 comp = (Z_RecordComposition *)
@@ -1847,6 +1894,25 @@ void Yaz_Proxy::recv_GDU(Z_GDU *apdu, int len)
         yaz_log (YLOG_LOG, "%sReceiving %s from client %d bytes",
                  m_session_str, gdu_name(apdu), len);
 
+#if 0
+    // try to make a _bad_ attribute set ID .. Don't enable this in prod.
+    if (apdu->which == Z_GDU_Z3950 
+        && apdu->u.z3950->which == Z_APDU_searchRequest)
+    {
+        Z_SearchRequest *req = apdu->u.z3950->u.searchRequest;
+        if (req->query && req->query->which == Z_Query_type_1)
+        {
+            Z_RPNQuery *rpnquery = req->query->u.type_1;
+            if (rpnquery->attributeSetId)
+            {
+                rpnquery->attributeSetId[0] = -2;
+                rpnquery->attributeSetId[1] = -1;
+                yaz_log(YLOG_WARN, "%sBAD FIXUP TEST", m_session_str);
+            }
+        }
+    }
+#endif
+
 #if HAVE_GETTIMEOFDAY
     gettimeofday((struct timeval *) m_time_tv, 0);
 #endif
@@ -1854,8 +1920,16 @@ void Yaz_Proxy::recv_GDU(Z_GDU *apdu, int len)
     m_pdu_stat.add_bytes(1);
 
     GDU *gdu = new GDU(apdu);
-    m_in_queue.enqueue(gdu);
 
+    if (gdu->get() == 0)
+    {
+        delete gdu;
+        yaz_log(YLOG_LOG, "%sUnable to encode package", m_session_str);
+        m_in_queue.clear();
+        dec_ref();
+        return;
+    }
+    m_in_queue.enqueue(gdu);
     recv_GDU_more(false);
 }
 
@@ -1981,7 +2055,10 @@ void Yaz_Proxy::recv_GDU_more(bool normal)
     while (m_timeout_mode == timeout_normal && (g = m_in_queue.dequeue()))
     {
         m_timeout_mode = timeout_busy;
+        inc_ref();
         recv_GDU_reduce(g);
+        if (dec_ref())
+            break;
     }
 }
 
@@ -3060,28 +3137,32 @@ void Yaz_Proxy::handle_init(Z_APDU *apdu)
 
             handle_charset_lang_negotiation(apdu2);
 
+           if (m_timeout_mode == timeout_busy)
+               m_timeout_mode = timeout_normal;
             send_to_client(apdu2);
-            m_timeout_mode = timeout_normal;
             return;
         }
     }
     m_client->m_init_flag = 1;
 
-#if USE_AUTH_MSG
-    Auth_Msg *m = new Auth_Msg;
-    m->m_proxy = this;
-    z_APDU(odr_encode(), &apdu, 0, "encode");
-    char *apdu_buf = odr_getbuf(odr_encode(), &m->m_apdu_len, 0);
-    m->m_apdu_buf = (char*) nmem_malloc(m->m_nmem, m->m_apdu_len);
-    memcpy(m->m_apdu_buf, apdu_buf, m->m_apdu_len);
-    odr_reset(odr_encode());
-
-    inc_ref();
-    m_my_thread->put(m);
-#else
-    int ret = handle_authentication(apdu);
-    result_authentication(apdu, ret);
-#endif
+    if (m_num_msg_threads && m_my_thread)
+    {
+        Auth_Msg *m = new Auth_Msg;
+        m->m_proxy = this;
+        z_APDU(odr_encode(), &apdu, 0, "encode");
+        char *apdu_buf = odr_getbuf(odr_encode(), &m->m_apdu_len, 0);
+        m->m_apdu_buf = (char*) nmem_malloc(m->m_nmem, m->m_apdu_len);
+        memcpy(m->m_apdu_buf, apdu_buf, m->m_apdu_len);
+        odr_reset(odr_encode());
+        
+        inc_ref();
+        m_my_thread->put(m);
+    }
+    else
+    {
+        int ret = handle_authentication(apdu);
+        result_authentication(apdu, ret);
+    }
 }
 
 void Yaz_Proxy::handle_incoming_Z_PDU(Z_APDU *apdu)
@@ -3239,17 +3320,8 @@ void Yaz_Proxy::releaseClient()
         m_parent->pre_init();
 }
 
-bool Yaz_Proxy::dec_ref(bool main_ptr)
+bool Yaz_Proxy::dec_ref()
 {
-    assert(m_ref_count > 0);
-    if (main_ptr)
-    {
-        yaz_log(YLOG_LOG, "%sdec_ref", m_session_str);
-        if (m_main_ptr_dec)
-            return false;
-        m_main_ptr_dec = true;
-    }
-
     m_http_keepalive = 0;
 
     --m_ref_count;
@@ -3276,8 +3348,8 @@ void Yaz_ProxyClient::shutdown()
 
     if (m_server)
     {
-        m_waiting = 1;      // ensure it's released from Proxy in releaseClient
-        m_server->dec_ref(true);
+        m_waiting = 1;   // ensure it's released from Yaz_Proxy::releaseClient
+        m_server->dec_ref();
     }
     else
         delete this;
@@ -3286,9 +3358,8 @@ void Yaz_ProxyClient::shutdown()
 void Yaz_Proxy::failNotify()
 {
     inc_request_no();
-    yaz_log (YLOG_LOG, "%sConnection closed by client",
-             get_session_str());
-    dec_ref(true);
+    yaz_log (YLOG_LOG, "%sConnection closed by client", get_session_str());
+    dec_ref();
 }
 
 void Yaz_Proxy::send_response_fail_client(const char *addr)
@@ -3489,7 +3560,7 @@ void Yaz_Proxy::timeoutNotify()
             inc_request_no();
             m_in_queue.clear();
             yaz_log (YLOG_LOG, "%sTimeout (client to proxy)", m_session_str);
-            dec_ref(true);
+            dec_ref();
             break;
         case timeout_reduce:
             timeout(m_client_idletime);
@@ -3639,7 +3710,9 @@ void Yaz_ProxyClient::recv_Z_PDU(Z_APDU *apdu, int len)
         *imv1 = '\0';
         if (imv0)
             strcat(imv1, imv0);
+#ifdef VERSION
         strcat(imv1, "/" VERSION);
+#endif
         ir->implementationVersion = imv1;
         
         // apply YAZ Proxy implementation name