NCBI C Toolkit Cross Reference

/*  $Id: hspfilter_queue.c,v 1.2 2024/01/03 20:26:33 ucko Exp $
 * ===========================================================================
 *
 *                            PUBLIC DOMAIN NOTICE
 *               National Center for Biotechnology Information
 *
 *  This software/database is a "United States Government Work" under the
 *  terms of the United States Copyright Act.  It was written as part of
 *  the author's official duties as a United States Government employee and
 *  thus cannot be copyrighted.  This software/database is freely available
 *  to the public for use. The National Library of Medicine and the U.S.
 *  Government have not placed any restriction on its use or reproduction.
 *
 *  Although all reasonable efforts have been taken to ensure the accuracy
 *  and reliability of the software and data, the NLM and the U.S.
 *  Government do not and cannot warrant the performance or results that
 *  may be obtained by using this software or data. The NLM and the U.S.
 *  Government disclaim all warranties, express or implied, including
 *  warranties of performance, merchantability or fitness for any particular
 *  purpose.
 *
 *  Please cite the author in any work or product based on this material.
 *
 * ===========================================================================
 *
 * Author:  Ning Ma
 *
 */

/** @file hspfilter_queue.c
 * Default implementation of the BlastHSPWriter interface to save hits from
 * a BLAST search, and subsequently return them in sorted order.
 */

#ifndef SKIP_DOXYGEN_PROCESSING
static char const rcsid[] = 
    "$Id: hspfilter_queue.c,v 1.2 2024/01/03 20:26:33 ucko Exp $";
#endif /* SKIP_DOXYGEN_PROCESSING */


#include <ncbithr.h>
#include <algo/blast/core/blast_hspstream.h>
#include <algo/blast/api/hspfilter_queue.h>
//#include <algo/blast/core/blast_util.h>

/** Data structure used by the writer */
typedef struct BlastHSPQueueData {
   ListNode * start;      /**< First element of the queue */
   ListNode * end;        /**< First element of the queue */
   Boolean    writeDone;  /**< Has writing to this stream been finished? */
   TNlmMutex  lock;       /**< reading/writing lock */
   TNlmSemaphore sema;    /**< Semaphore for reading */
} BlastHSPQueueData;

/*************************************************************/
/** The following are implementations for BlastHSPWriter ADT */

/** Perform pre-run stage-specific initialization 
 * @param data The internal data structure [in][out]
 * @param results The HSP results to operate on  [in]
 */ 
static int 
s_BlastHSPQueueInit(void* data, BlastHSPResults* results)
{
   BlastHSPQueueData * q_data = data;
   return 0;
}

/** Perform post-run clean-ups
 * @param data The buffered data structure [in]
 * @param results The HSP results to propagate [in][out]
 */ 
static int 
s_BlastHSPQueueFinal(void* data, BlastHSPResults* results)
{
   BlastHSPQueueData * q_data = data;

   NlmMutexLockEx(&q_data->lock);
   q_data->writeDone = TRUE;
   NlmSemaPost(q_data->sema);
   NlmMutexUnlock(q_data->lock);

   return 0;
}

/** Perform writing task
 * ownership of the HSP list and sets the dereferenced pointer to NULL.
 * @param data To store results to [in][out]
 * @param hsp_list Pointer to the HSP list to save in the queue. [in]
 */
static int 
s_BlastHSPQueueRun(void* data, BlastHSPList* hsp_list)
{
   BlastHSPQueueData * q_data = data;

   if (!hsp_list)
      return 0;

   if (hsp_list->hspcnt == 0) {
      Blast_HSPListFree(hsp_list);
      return 0;
   }

   if (q_data->writeDone)
      return -1;

   NlmMutexLockEx(&q_data->lock);
   q_data->end = ListNodeAddPointer(&q_data->end, 0, (void *)hsp_list);
   if (!q_data->start)
      q_data->start = q_data->end;
   hsp_list = NULL;
   NlmSemaPost(q_data->sema);
   NlmMutexUnlock(q_data->lock);

   return 0; 
}

/** Free the writer 
 * @param writer The writer to free [in]
 * @return NULL.
 */
static
BlastHSPWriter*
s_BlastHSPQueueFree(BlastHSPWriter* writer) 
{
   ListNode * p;
   BlastHSPQueueData *q_data = writer->data;

   NlmSemaDestroy(q_data->sema);
   NlmMutexDestroy(q_data->lock);

   for(p = q_data->start; p; p = p->next) {
      p->ptr = (void *) Blast_HSPListFree((BlastHSPList*) p->ptr);
   }
   q_data->start = ListNodeFree(q_data->start);
   sfree(writer->data);
   sfree(writer);
   return NULL;
}

/** create the writer
 * @param params Pointer to the hit paramters [in]
 * @param query_info BlastQueryInfo (not used) [in]
 * @return writer
 */
static
BlastHSPWriter* 
s_BlastHSPQueueNew(void* params, BlastQueryInfo* query_info)
{
   BlastHSPWriter * writer = NULL;
   BlastHSPQueueData * data = NULL;

   /* allocate space for writer */
   writer = malloc(sizeof(BlastHSPWriter));

   /* fill up the function pointers */
   writer->InitFnPtr   = (BlastHSPWriterInitFn)  &s_BlastHSPQueueInit;
   writer->FinalFnPtr  = (BlastHSPWriterFinalFn) &s_BlastHSPQueueFinal;
   writer->FreeFnPtr   = &s_BlastHSPQueueFree;
   writer->RunFnPtr    = &s_BlastHSPQueueRun;

   /* allocate for data structure */
   writer->data = calloc(1,sizeof(BlastHSPQueueData));
   data = writer->data;
   data->sema = NlmSemaInit(0);
    
   return writer;
}

/*************************************************************/
/** The following are exported functions to be used by APP   */

BlastHSPQueueParams*
BlastHSPQueueParamsNew()
{
    return NULL;
}

BlastHSPQueueParams*
BlastHSPQueueParamsFree(BlastHSPQueueParams* opts)
{
    return NULL;
}

BlastHSPWriterInfo*
BlastHSPQueueInfoNew(BlastHSPQueueParams* params) {
   BlastHSPWriterInfo * writer_info =
                        malloc(sizeof(BlastHSPWriterInfo)); 
   writer_info->NewFnPtr = (BlastHSPWriterNewFn) &s_BlastHSPQueueNew;
   writer_info->params = params;
   return writer_info;
}

/************************************************************/
/** The follwoing is added to support queue implementation  */

int BlastHSPQueueRead(void* data, BlastHSPList** hsp_list_out) 
{
   BlastHSPQueueData* q_data = (BlastHSPQueueData*) data;
   int status = kBlastHSPStream_Error;

   /* Lock the mutex */
   NlmMutexLockEx(&q_data->lock);

   if (!q_data->writeDone) {
      while (!q_data->writeDone && !q_data->start) {
         /* Decrement the semaphore count to 0, then wait for it to be 
          * incremented. Note that mutex must be locked whenever the 
          * contents of the stream are checked, but it must be unlocked
          * for the semaphore wait. */
         NlmMutexUnlock(q_data->lock);
         NlmSemaWait(q_data->sema);
         NlmMutexLockEx(&q_data->lock);
      }
   }

   if (!q_data->start) {
      /* Nothing in the queue, but no more writing to the queue is expected. */
      *hsp_list_out = NULL;
      status =  kBlastHSPStream_Eof;
   } else {
      ListNode* start_node = q_data->start;

      *hsp_list_out = (BlastHSPList*) start_node->ptr;

      q_data->start = start_node->next;
      start_node->next = NULL;
      ListNodeFree(start_node);
      if (!q_data->start)
         q_data->end = NULL;
      status = kBlastHSPStream_Success;
   }

   NlmMutexUnlock(q_data->lock);

   return status;
}