ZEND_API zval *_zend_ts_hash_index_add_or_update(TsHashTable *ht, zend_ulong h, zval *pData, int flag ZEND_FILE_LINE_DC)
{
	zval *retval;

	begin_write(ht);
	retval = _zend_hash_index_add_or_update(TS_HASH(ht), h, pData, flag ZEND_FILE_LINE_RELAY_CC);
	end_write(ht);

	return retval;
}

ZEND_API int zend_ts_hash_del(TsHashTable *ht, zend_string *key)
{
	int retval;

	begin_write(ht);
	retval = zend_hash_del(TS_HASH(ht), key);
	end_write(ht);

	return retval;
}

ZEND_API void zend_ts_hash_apply_with_arguments(TsHashTable *ht, apply_func_args_t apply_func, int num_args, ...)
{
	va_list args;

	va_start(args, num_args);
	begin_write(ht);
	zend_hash_apply_with_arguments(TS_HASH(ht), apply_func, num_args, args);
	end_write(ht);
	va_end(args);
}

ZEND_API int zend_ts_hash_rehash(TsHashTable *ht)
{
	int retval;

	begin_write(ht);
	retval = zend_hash_rehash(TS_HASH(ht));
	end_write(ht);

	return retval;
}

ZEND_API int zend_ts_hash_index_del(TsHashTable *ht, zend_ulong h)
{
	int retval;

	begin_write(ht);
	retval = zend_hash_index_del(TS_HASH(ht), h);
	end_write(ht);

	return retval;
}

ZEND_API int zend_ts_hash_sort(TsHashTable *ht, sort_func_t sort_func, compare_func_t compare_func, int renumber)
{
	int retval;

	begin_write(ht);
	retval = zend_hash_sort_ex(TS_HASH(ht), sort_func, compare_func, renumber);
	end_write(ht);

	return retval;
}

ZEND_API int _zend_ts_hash_index_update_or_next_insert(TsHashTable *ht, ulong h, void *pData, uint nDataSize, void **pDest, int flag ZEND_FILE_LINE_DC)
{
	int retval;

	begin_write(ht);
	retval = _zend_hash_index_update_or_next_insert(TS_HASH(ht), h, pData, nDataSize, pDest, flag ZEND_FILE_LINE_RELAY_CC);
	end_write(ht);

	return retval;
}

ZEND_API int zend_ts_hash_add_empty_element(TsHashTable *ht, char *arKey, uint nKeyLength)
{
	int retval;

	begin_write(ht);
	retval = zend_hash_add_empty_element(TS_HASH(ht), arKey, nKeyLength);
	end_write(ht);

	return retval;
}

ZEND_API int _zend_ts_hash_add_or_update(TsHashTable *ht, char *arKey, uint nKeyLength, void *pData, uint nDataSize, void **pDest, int flag ZEND_FILE_LINE_DC)
{
	int retval;

	begin_write(ht);
	retval = _zend_hash_add_or_update(TS_HASH(ht), arKey, nKeyLength, pData, nDataSize, pDest, flag ZEND_FILE_LINE_RELAY_CC);
	end_write(ht);

	return retval;
}

/* Join a group.  The group context is returned in *contextp.  
 */
hot_err_t hot_ens_Join(
    hot_ens_JoinOps_t *jops,
    hot_context_t *contextp	/*OUT*/ 
) {
    hot_err_t err = HOT_OK ;
    hot_context_t s ;

    /* Initialize global state if not done so already.
     */
    if (!g.initialized) {
	err = hot_ens_Init(jops->outboard, jops->argv); 
	if (err != HOT_OK)
	    return err;
    }

    begin_write(); {    
	begin_critical(); {    
	    /* Allocate a new group context 
	     * Initialize the group record.
	     */
	    s = alloc_context();
	    s->joining = 1 ;
	    s->leaving = 0 ;
	    s->conf = jops->conf;
	    s->env = jops->env;
	    s->view = NULL ;
	    *contextp = s ;
	} end_critical();
	
	/* Write the downcall.
	 */
  	write_hdr(s,DN_JOIN);
	write_int(jops->heartbeat_rate);
	write_string(jops->transports);
	write_string(jops->protocol);
	write_string(jops->group_name);
	write_string(jops->properties);
	write_bool(jops->use_properties);
	write_bool(jops->groupd);
	write_string(jops->params);
	write_bool(jops->client);
	write_bool(jops->debug);

	if (jops->endpt.name[0] != 0x0) {
	    hot_sys_Warning("HOT_OUTBOARD does not support 'endpt' in join ops") ;
	    jops->endpt.name[0] = 0x0;
	}
	write_string(jops->princ);
	write_string(jops->key);
	write_bool(jops->secure);

    } end_write();

    return HOT_OK;
}

ZEND_API void zend_ts_hash_destroy(TsHashTable *ht)
{
	begin_write(ht);
	zend_hash_destroy(TS_HASH(ht));
	end_write(ht);

#ifdef ZTS
	tsrm_mutex_free(ht->mx_reader);
	tsrm_mutex_free(ht->mx_writer);
#endif
}

/* Inform Ensemble that the state-transfer is complete. 
 */
hot_err_t hot_ens_XferDone(
    hot_context_t s
) {
    begin_write(); {
	begin_critical(); {
	    if (s->leaving) {
		hot_sys_Panic("hot_ens_XferDone: member is leaving") ;
	    }
	} end_critical();
	write_hdr(s,DN_XFERDONE);
    } end_write();
    return HOT_OK;
}

/* Request a new view to be installed.
 */
hot_err_t hot_ens_RequestNewView(
    hot_context_t s
) {
    begin_write(); {
	begin_critical(); {
	    if (s->leaving) {
		hot_sys_Panic("hot_ens_RequestNewView: member is leaving") ;
	    }
	} end_critical();
	write_hdr(s,DN_PROMPT);
    } end_write();
    return HOT_OK;
}

/* Request a Rekey operation.
 */
hot_err_t hot_ens_Rekey(
    hot_context_t s
) {
    begin_write(); {
	begin_critical(); {
	    if (s->leaving) {
		hot_sys_Panic("hot_ens_Rekey: member is leaving") ;
	    }
	} end_critical();
	write_hdr(s,DN_REKEY);
    } end_write();
    return HOT_OK;
}

// Releases the reader_writer_lock
void reader_writer_lock::unlock() {
    if( my_current_writer!=tbb_thread::id() ) {
        // A writer owns the lock
        __TBB_ASSERT(is_current_writer(), "caller of reader_writer_lock::unlock() does not own the lock.");
        __TBB_ASSERT(writer_head, NULL);
        __TBB_ASSERT(writer_head->status==active, NULL);
        scoped_lock *a_writer_lock = writer_head;
        end_write(a_writer_lock);
        __TBB_ASSERT(a_writer_lock != writer_head, "Internal error: About to turn writer_head into dangling reference.");
        delete a_writer_lock;
    } else {
        end_read();
    }
}

static void outb_atexit(void) {
  char eof = EOF;

  /* Send a kill signal to the child.
   */
  kill(cid, SIGKILL);
  waitpid(cid, NULL, 0);
  
  /* In case signal fails, write EOF to the pipe.
   */
  begin_write() ;
  do_write(&eof, 4);
  end_write() ;
}

/* zodiac_spew - Takes a message type, an array of data words, and a length
 * for the array, and prepends a 5 word header (including checksum).
 * The data words are expected to be checksummed.
 */
static ssize_t zodiac_spew(struct gps_device_t *session, unsigned short type,
			   unsigned short *dat, int dlen)
{
    struct header h;
    int i;
    char buf[BUFSIZ];

    h.sync = 0x81ff;
    h.id = (unsigned short)type;
    h.ndata = (unsigned short)(dlen - 1);
    h.flags = 0;
    h.csum = zodiac_checksum((unsigned short *)&h, 4);

    if (!BAD_SOCKET(session->gpsdata.gps_fd)) {
	size_t hlen, datlen;
	hlen = sizeof(h);
	datlen = sizeof(unsigned short) * dlen;
	if (end_write(session->gpsdata.gps_fd, &h, hlen) != (ssize_t) hlen ||
	    end_write(session->gpsdata.gps_fd, dat,
		      datlen) != (ssize_t) datlen) {
	    gpsd_log(&session->context->errout, LOG_RAW,
		     "Reconfigure write failed\n");
	    return -1;
	}
    }

    (void)snprintf(buf, sizeof(buf),
		   "%04x %04x %04x %04x %04x",
		   h.sync, h.id, h.ndata, h.flags, h.csum);
    for (i = 0; i < dlen; i++)
	str_appendf(buf, sizeof(buf), " %04x", dat[i]);

    gpsd_log(&session->context->errout, LOG_RAW,
	     "Sent Zodiac packet: %s\n", buf);

    return 0;
}

struct linked_list* list_remove(struct linked_list_head *list, int val) {
  struct linked_list **p, *ret=NULL;
  begin_write(list->sync);
  p=&list->head;
  while (*p) {
    if ((*p)->nb==val) {
      ret=*p;
      *p=(*p)->next;
      break;
    }
    p=&(*p)->next;
  }
  end_write(list->sync);
  return ret;
}

struct linked_list* remove_element(struct linked_list_head *list, int val) {
	struct linked_list **p, *ret = NULL;
	begin_write(&list->sync);
	sleep(rand()%3 + 1);
	p = &list->head;
	while((*p) != NULL) {
		if ((*p)->nb == val) {
			ret = *p;
			*p = (*p)->next;
			break;
		}
		p = &(*p)->next;
	}
	end_write(&list->sync);
	return ret;
}

/* Report group members as failure-suspected.
 * 
 * NB:  In the initial implementation, this downcall will not be supported.
 *      (if invoked, an exeption will be raised by OCAML).
 */
hot_err_t hot_ens_Suspect(
    hot_context_t s,
    hot_endpt_t *suspects, 
    int nsuspects
) {
    begin_write(); {
	begin_critical(); {
	    if (s->leaving) {
		hot_sys_Panic("hot_ens_Suspect: member is leaving") ;
	    }
	} end_critical();
	write_hdr(s,DN_SUSPECT);
	write_endpList(nsuspects, suspects);
    } end_write();
    return HOT_OK;
}