fixed #1804 RT write stall by slow search queries

git-svn-id: svn://svn.sphinxsearch.com/sphinx/trunk@4627 406a0c4d-033a-0410-8de8-e80135713968

config/config.h.in

@@ -160,7 +160,7 @@
 #undef HAVE_STRTOL
 
 /* Whether we have internal atomic functions */
-#undef HAVE_SYNC_FETCH_AND_ADD
+#undef HAVE_SYNC_FETCH
 
 /* Define to 1 if you have the <sys/epoll.h> header file. */
 #undef HAVE_SYS_EPOLL_H

config/config_cmake.h.in

@@ -160,7 +160,7 @@
 #cmakedefine HAVE_STRTOL ${HAVE_STRTOL}
 
 /* Whether we have internal atomic functions */
-#cmakedefine HAVE_SYNC_FETCH_AND_ADD
+#cmakedefine HAVE_SYNC_FETCH
 
 /* Define to 1 if you have the <sys/epoll.h> header file. */
 #cmakedefine HAVE_SYS_EPOLL_H ${HAVE_SYS_EPOLL_H}

configure

@@ -6325,7 +6325,7 @@ done
 
 
 
-# check for interlocked increment in linux
+# check for interlocked increment and decrement in linux
 
 interlocked_prog="
 
@@ -6333,6 +6333,7 @@ int interlocked_routine ( )
 {
 	volatile int ia=0;
     __sync_fetch_and_add( &ia, 1 ); // atomic increment
+	__sync_fetch_and_sub( &ia, 1 ); // atomic decrement
 }
 
 int main ()
@@ -6384,7 +6385,7 @@ $as_echo "$sphinx_cv_interlocked" >&6; }
 # final check
 if test x$sphinx_cv_interlocked = xyes; then
 
-$as_echo "#define HAVE_SYNC_FETCH_AND_ADD 1" >>confdefs.h
+$as_echo "#define HAVE_SYNC_FETCH 1" >>confdefs.h
 
 fi
 

configure.ac

@@ -150,7 +150,7 @@ AC_SEARCH_LIBS([dlopen],[dl dld],[],[])
 AC_CHECK_FUNCS(dlopen dlerror)
 
 
-# check for interlocked increment in linux
+# check for interlocked increment and decrement in linux
 
 dnl helper that runs a test program and checks for success
 interlocked_prog="
@@ -159,6 +159,7 @@ int interlocked_routine ( )
 {
 	volatile int ia=0;
     __sync_fetch_and_add( &ia, 1 ); // atomic increment
+	__sync_fetch_and_sub( &ia, 1 ); // atomic decrement
 }
 
 int main ()
@@ -178,7 +179,7 @@ AC_CACHE_CHECK([for interlocked builtins], [sphinx_cv_interlocked],
 
 # final check
 if test x$sphinx_cv_interlocked = xyes; then
-    AC_DEFINE(HAVE_SYNC_FETCH_AND_ADD,1,[Whether we have internal atomic functions])
+    AC_DEFINE(HAVE_SYNC_FETCH,1,[Whether we have internal atomic functions])
 fi
 
 dnl --------------------------------------------------------------------------

src/searchd.cpp

@@ -16914,8 +16914,8 @@ void HandleMysqlAttach ( SqlRowBuffer_c & tOut, const SqlStmt_t & tStmt )
 	}
 
 	ISphRtIndex * pRtTo = (ISphRtIndex*)pTo->m_pIndex;
-
-	if ( !pRtTo->AttachDiskIndex ( pFrom->m_pIndex, sError ) )
+	bool bOk = pRtTo->AttachDiskIndex ( pFrom->m_pIndex, sError );
+	if ( !bOk )
 	{
 		pFrom->Unlock();
 		pTo->Unlock();
@@ -17374,8 +17374,9 @@ static void ModifyIndexAttrs ( const ServedIndex_t * pLocal, bool bAdd, CSphStri
 		return;
 
 	if ( pLocal->m_pIndex->IsRT() )
+	{
 		pLocal->m_pIndex->AddRemoveAttribute ( bAdd, sAttrName, eAttrType, iPos, sError );
-	else
+	} else
 	{
 		if ( RenameWithRollback ( pLocal ) )
 			pLocal->m_pIndex->AddRemoveAttribute ( bAdd, sAttrName, eAttrType, iPos, sError );

src/sphinx.cpp

@@ -1549,8 +1549,8 @@ public:
 	virtual bool				Merge ( CSphIndex * pSource, const CSphVector<CSphFilterSettings> & dFilters, bool bMergeKillLists );
 
 	template <class QWORDDST, class QWORDSRC>
-	static bool					MergeWords ( const CSphIndex_VLN * pDstIndex, const CSphIndex_VLN * pSrcIndex, const ISphFilter * pFilter, const CSphVector<SphDocID_t> & dKillList, SphDocID_t uMinID, CSphHitBuilder * pHitBuilder, CSphString & sError, CSphSourceStats & tStat, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pForceTerminate );
-	static bool					DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_VLN * pSrcIndex, bool bMergeKillLists, ISphFilter * pFilter, const CSphVector<SphDocID_t> & dKillList, CSphString & sError, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pForceTerminate );
+	static bool					MergeWords ( const CSphIndex_VLN * pDstIndex, const CSphIndex_VLN * pSrcIndex, const ISphFilter * pFilter, const CSphVector<SphDocID_t> & dKillList, SphDocID_t uMinID, CSphHitBuilder * pHitBuilder, CSphString & sError, CSphSourceStats & tStat, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pGlobalStop, volatile bool * pLocalStop );
+	static bool					DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_VLN * pSrcIndex, bool bMergeKillLists, ISphFilter * pFilter, const CSphVector<SphDocID_t> & dKillList, CSphString & sError, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pGlobalStop, volatile bool * pLocalStop );
 
 	virtual int					UpdateAttributes ( const CSphAttrUpdate & tUpd, int iIndex, CSphString & sError, CSphString & sWarning );
 	virtual bool				SaveAttributes ( CSphString & sError ) const;
@@ -13748,7 +13748,7 @@ int CSphIndex_VLN::Build ( const CSphVector<CSphSource*> & dSources, int iMemory
 /////////////////////////////////////////////////////////////////////////////
 
 
-static bool CopyFile ( const char * sSrc, const char * sDst, CSphString & sErrStr, ThrottleState_t * pThrottle, volatile bool * pForceTerminate )
+static bool CopyFile ( const char * sSrc, const char * sDst, CSphString & sErrStr, ThrottleState_t * pThrottle, volatile bool * pGlobalStop, volatile bool * pLocalStop )
 {
 	assert ( sSrc );
 	assert ( sDst );
@@ -13771,7 +13771,7 @@ static bool CopyFile ( const char * sSrc, const char * sDst, CSphString & sErrSt
 
 		while ( iFileSize > 0 )
 		{
-			if ( *pForceTerminate )
+			if ( *pGlobalStop || *pLocalStop )
 				return false;
 
 			DWORD iSize = (DWORD) Min ( iFileSize, (SphOffset_t)iBufSize );
@@ -14146,14 +14146,14 @@ public:
 	template < typename QWORD >
 	inline void TransferData ( QWORD & tQword, SphWordID_t iWordID, const BYTE * sWord,
 							const CSphIndex_VLN * pSourceIndex, const ISphFilter * pFilter,
-							const CSphVector<SphDocID_t> & dKillList, volatile bool * pForceTerminate )
+							const CSphVector<SphDocID_t> & dKillList, volatile bool * pGlobalStop, volatile bool * pLocalStop )
 	{
 		CSphAggregateHit tHit;
 		tHit.m_uWordID = iWordID;
 		tHit.m_sKeyword = sWord;
 		tHit.m_dFieldMask.UnsetAll();
 
-		while ( CSphMerger::NextDocument ( tQword, pSourceIndex, pFilter, dKillList ) && !*pForceTerminate )
+		while ( CSphMerger::NextDocument ( tQword, pSourceIndex, pFilter, dKillList ) && !*pGlobalStop && !*pLocalStop )
 		{
 			if ( tQword.m_bHasHitlist )
 				TransferHits ( tQword, tHit );
@@ -14207,7 +14207,7 @@ template < typename QWORDDST, typename QWORDSRC >
 bool CSphIndex_VLN::MergeWords ( const CSphIndex_VLN * pDstIndex, const CSphIndex_VLN * pSrcIndex,
 								const ISphFilter * pFilter, const CSphVector<SphDocID_t> & dKillList, SphDocID_t uMinID,
 								CSphHitBuilder * pHitBuilder, CSphString & sError, CSphSourceStats & tStat,
-								CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pForceTerminate )
+								CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pGlobalStop, volatile bool * pLocalStop )
 {
 	CSphAutofile tDummy;
 	pHitBuilder->CreateIndexFiles ( pDstIndex->GetIndexFileName("tmp.spd").cstr(),
@@ -14248,7 +14248,7 @@ bool CSphIndex_VLN::MergeWords ( const CSphIndex_VLN * pDstIndex, const CSphInde
 	tDstDocs.Open ( pDstIndex->GetIndexFileName("spd"), SPH_O_READ, sError );
 	tDstHits.Open ( pDstIndex->GetIndexFileName("spp"), SPH_O_READ, sError );
 
-	if ( !sError.IsEmpty() || *pForceTerminate )
+	if ( !sError.IsEmpty() || *pGlobalStop || *pLocalStop )
 		return false;
 
 	int iDstInlineSize = pDstIndex->m_tSettings.m_eDocinfo==SPH_DOCINFO_INLINE ? pDstIndex->m_tSchema.GetRowSize() : 0;
@@ -14282,7 +14282,7 @@ bool CSphIndex_VLN::MergeWords ( const CSphIndex_VLN * pDstIndex, const CSphInde
 			iWords = 0;
 		}
 
-		if ( *pForceTerminate )
+		if ( *pGlobalStop || *pLocalStop )
 			return false;
 
 		const int iCmp = tDstReader.CmpWord ( tSrcReader );
@@ -14291,14 +14291,14 @@ bool CSphIndex_VLN::MergeWords ( const CSphIndex_VLN * pDstIndex, const CSphInde
 		{
 			// transfer documents and hits from destination
 			CSphMerger::PrepareQword<QWORDDST> ( tDstQword, tDstReader, uDstMinID, bWordDict );
-			tMerger.TransferData<QWORDDST> ( tDstQword, tDstReader.m_uWordID, tDstReader.GetWord(), pDstIndex, pFilter, dKillList, pForceTerminate );
+			tMerger.TransferData<QWORDDST> ( tDstQword, tDstReader.m_uWordID, tDstReader.GetWord(), pDstIndex, pFilter, dKillList, pGlobalStop, pLocalStop );
 			bDstWord = tDstReader.Read();
 
 		} else if ( !bDstWord || ( bSrcWord && iCmp>0 ) )
 		{
 			// transfer documents and hits from source
 			CSphMerger::PrepareQword<QWORDSRC> ( tSrcQword, tSrcReader, uSrcMinID, bWordDict );
-			tMerger.TransferData<QWORDSRC> ( tSrcQword, tSrcReader.m_uWordID, tSrcReader.GetWord(), pSrcIndex, NULL, CSphVector<SphDocID_t>(), pForceTerminate );
+			tMerger.TransferData<QWORDSRC> ( tSrcQword, tSrcReader.m_uWordID, tSrcReader.GetWord(), pSrcIndex, NULL, CSphVector<SphDocID_t>(), pGlobalStop, pLocalStop );
 			bSrcWord = tSrcReader.Read();
 
 		} else // merge documents and hits inside the word
@@ -14328,7 +14328,7 @@ bool CSphIndex_VLN::MergeWords ( const CSphIndex_VLN * pDstIndex, const CSphInde
 
 			while ( bDstDocs || bSrcDocs )
 			{
-				if ( *pForceTerminate )
+				if ( *pGlobalStop || *pLocalStop )
 					return false;
 
 				if ( !bSrcDocs || ( bDstDocs && tDstQword.m_tDoc.m_uDocID < tSrcQword.m_tDoc.m_uDocID ) )
@@ -14455,15 +14455,16 @@ bool CSphIndex_VLN::Merge ( CSphIndex * pSource, const CSphVector<CSphFilterSett
 	dKillList[0] = 0;
 	dKillList.Last() = DOCID_MAX;
 
-	bool bForceTerminate = false;
+	bool bGlobalStop = false;
+	bool bLocalStop = false;
 	return CSphIndex_VLN::DoMerge ( this, (const CSphIndex_VLN *)pSource, bMergeKillLists, pFilter.Ptr(),
-									dKillList, m_sLastError, m_tProgress, &g_tThrottle, &bForceTerminate );
+									dKillList, m_sLastError, m_tProgress, &g_tThrottle, &bGlobalStop, &bLocalStop );
 }
 
 bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_VLN * pSrcIndex,
 							bool bMergeKillLists, ISphFilter * pFilter, const CSphVector<SphDocID_t> & dKillList
 							, CSphString & sError, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle,
-							volatile bool * pForceTerminate )
+							volatile bool * pGlobalStop, volatile bool * pLocalStop )
 {
 	assert ( pDstIndex && pSrcIndex );
 
@@ -14568,7 +14569,7 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 		CSphMatch tMatch;
 		while ( iSrcCount < pSrcIndex->m_iDocinfo || iDstCount < pDstIndex->m_iDocinfo )
 		{
-			if ( *pForceTerminate )
+			if ( *pGlobalStop || *pLocalStop )
 				return false;
 
 			SphDocID_t iDstDocID, iSrcDocID;
@@ -14682,7 +14683,7 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 		CSphString sSrc = !pDstIndex->m_bIsEmpty ? pDstIndex->GetIndexFileName("spa") : pSrcIndex->GetIndexFileName("spa");
 		CSphString sDst = pDstIndex->GetIndexFileName("tmp.spa");
 
-		if ( !CopyFile ( sSrc.cstr(), sDst.cstr(), sError, pThrottle, pForceTerminate ) )
+		if ( !CopyFile ( sSrc.cstr(), sDst.cstr(), sError, pThrottle, pGlobalStop, pLocalStop ) )
 			return false;
 
 	} else
@@ -14704,7 +14705,7 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 	CSphAutofile tTmpDict ( pDstIndex->GetIndexFileName("tmp8.spi"), SPH_O_NEW, sError, true );
 	CSphAutofile tDict ( pDstIndex->GetIndexFileName("tmp.spi"), SPH_O_NEW, sError );
 
-	if ( !sError.IsEmpty() || tTmpDict.GetFD()<0 || tDict.GetFD()<0 || *pForceTerminate )
+	if ( !sError.IsEmpty() || tTmpDict.GetFD()<0 || tDict.GetFD()<0 || *pGlobalStop || *pLocalStop )
 		return false;
 
 	CSphScopedPtr<CSphDict> pDict ( pDstIndex->m_pDict->Clone() );
@@ -14732,7 +14733,7 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 		{
 			if ( !CSphIndex_VLN::MergeWords < QwordDst, QwordSrc > ( pDstIndex, pSrcIndex, pFilter, dPhantomKiller,
 																	uMinDocid, &tHitBuilder, sError, tBuildHeader,
-																	tProgress, pThrottle, pForceTerminate ) )
+																	tProgress, pThrottle, pGlobalStop, pLocalStop ) )
 				return false;
 		} ) );
 	} else
@@ -14742,7 +14743,7 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 		{
 			if ( !CSphIndex_VLN::MergeWords < QwordDst, QwordSrc > ( pDstIndex, pSrcIndex, pFilter, dPhantomKiller
 																	, uMinDocid, &tHitBuilder, sError, tBuildHeader,
-																	tProgress,	pThrottle, pForceTerminate ) )
+																	tProgress,	pThrottle, pGlobalStop, pLocalStop ) )
 				return false;
 		} ) );
 	}
@@ -14766,7 +14767,7 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 
 		tBuildHeader.m_uKillListSize = dKillList.GetLength ();
 
-		if ( *pForceTerminate )
+		if ( *pGlobalStop || *pLocalStop )
 			return false;
 
 		if ( dKillList.GetLength() )
@@ -14778,7 +14779,7 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 
 	tKillList.Close ();
 
-	if ( *pForceTerminate )
+	if ( *pGlobalStop || *pLocalStop )
 		return false;
 
 	// finalize
@@ -14808,12 +14809,12 @@ bool CSphIndex_VLN::DoMerge ( const CSphIndex_VLN * pDstIndex, const CSphIndex_V
 
 bool sphMerge ( const CSphIndex * pDst, const CSphIndex * pSrc, const CSphVector<SphDocID_t> & dKillList,
 				CSphString & sError, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle,
-				volatile bool * pForceTerminate )
+				volatile bool * pGlobalStop, volatile bool * pLocalStop )
 {
 	const CSphIndex_VLN * pDstIndex = (const CSphIndex_VLN *)pDst;
 	const CSphIndex_VLN * pSrcIndex = (const CSphIndex_VLN *)pSrc;
 
-	return CSphIndex_VLN::DoMerge ( pDstIndex, pSrcIndex, false, NULL, dKillList, sError, tProgress, pThrottle, pForceTerminate );
+	return CSphIndex_VLN::DoMerge ( pDstIndex, pSrcIndex, false, NULL, dKillList, sError, tProgress, pThrottle, pGlobalStop, pLocalStop );
 }
 
 
@@ -18244,7 +18245,7 @@ XQNode_t * sphExpandXQNode ( XQNode_t * pNode, ExpansionContext_t & tCtx )
 	if ( !iWilds || iWilds==iLen )
 		return pNode;
 
-	ISphWordlist::Args_t tWordlist ( tCtx.m_bMergeSingles, tCtx.m_iExpansionLimit, tCtx.m_bHasMorphology, tCtx.m_eHitless );
+	ISphWordlist::Args_t tWordlist ( tCtx.m_bMergeSingles, tCtx.m_iExpansionLimit, tCtx.m_bHasMorphology, tCtx.m_eHitless, tCtx.m_pIndexData );
 
 	if ( !sphIsWild(*sFull) || tCtx.m_iMinInfixLen==0 )
 	{
@@ -18380,6 +18381,7 @@ ExpansionContext_t::ExpansionContext_t()
 	, m_bMergeSingles ( false )
 	, m_pPayloads ( NULL )
 	, m_eHitless ( SPH_HITLESS_NONE )
+	, m_pIndexData ( NULL )
 {}
 
 
@@ -31174,11 +31176,12 @@ const BYTE * CWordlist::AcquireDict ( const CSphWordlistCheckpoint * pCheckpoint
 }
 
 
-ISphWordlist::Args_t::Args_t ( bool bPayload, int iExpansionLimit, bool bHasMorphology, ESphHitless eHitless )
+ISphWordlist::Args_t::Args_t ( bool bPayload, int iExpansionLimit, bool bHasMorphology, ESphHitless eHitless, const void * pIndexData )
 	: m_bPayload ( bPayload )
 	, m_iExpansionLimit ( iExpansionLimit )
 	, m_bHasMorphology ( bHasMorphology )
 	, m_eHitless ( eHitless )
+	, m_pIndexData ( pIndexData )
 {
 	m_sBuf.Reserve ( 2048 * SPH_MAX_WORD_LEN * 3 );
 	m_dExpanded.Reserve ( 2048 );

src/sphinxint.h

@@ -438,7 +438,7 @@ public:
 	CSphVector<CSphAttrLocator>				m_dOverrideIn;
 	CSphVector<CSphAttrLocator>				m_dOverrideOut;
 
-	void *									m_pIndexData;			///< backend specific data
+	const void *							m_pIndexData;			///< backend specific data
 	CSphQueryProfile *						m_pProfile;
 	const SmallStringHash_T<int64_t> *		m_pLocalDocs;
 	int64_t									m_iTotalDocs;
@@ -1618,7 +1618,7 @@ void			sphColumnToLowercase ( char * sVal );
 bool			sphCheckQueryHeight ( const struct XQNode_t * pRoot, CSphString & sError );
 void			sphTransformExtendedQuery ( XQNode_t ** ppNode, const CSphIndexSettings & tSettings, bool bHasBooleanOptimization, const ISphKeywordsStat * pKeywords );
 void			TransformAotFilter ( XQNode_t * pNode, const CSphWordforms * pWordforms, const CSphIndexSettings& tSettings );
-bool			sphMerge ( const CSphIndex * pDst, const CSphIndex * pSrc, const CSphVector<SphDocID_t> & dKillList, CSphString & sError, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pForceTerminate );
+bool			sphMerge ( const CSphIndex * pDst, const CSphIndex * pSrc, const CSphVector<SphDocID_t> & dKillList, CSphString & sError, CSphIndexProgress & tProgress, ThrottleState_t * pThrottle, volatile bool * pGlobalStop, volatile bool * pLocalStop );
 CSphString		sphReconstructNode ( const XQNode_t * pNode, const CSphSchema * pSchema );
 
 void			sphSetUnlinkOld ( bool bUnlink );
@@ -1786,8 +1786,9 @@ public:
 		ISphSubstringPayload *		m_pPayload;
 		int							m_iTotalDocs;
 		int							m_iTotalHits;
+		const void *				m_pIndexData;
 
-		Args_t ( bool bPayload, int iExpansionLimit, bool bHasMorphology, ESphHitless eHitless );
+		Args_t ( bool bPayload, int iExpansionLimit, bool bHasMorphology, ESphHitless eHitless, const void * pIndexData );
 		~Args_t ();
 		void AddExpanded ( const BYTE * sWord, int iLen, int iDocs, int iHits );
 		const char * GetWordExpanded ( int iIndex ) const;
@@ -1830,6 +1831,7 @@ struct ExpansionContext_t
 	bool m_bMergeSingles;
 	CSphScopedPayload * m_pPayloads;
 	ESphHitless m_eHitless;
+	const void * m_pIndexData;
 
 	ExpansionContext_t ();
 };

src/sphinxstd.cpp

@@ -1818,6 +1818,10 @@ long CSphAtomic<long>::Inc()
 {
 	return InterlockedIncrement ( &m_uValue )-1;
 }
+long CSphAtomic<long>::Dec()
+{
+	return InterlockedDecrement  ( &m_uValue )+1;
+}
 #endif
 
 // fast check if we are built with right endianess settings

src/sphinxstd.h

@@ -1227,6 +1227,13 @@ public:
 		Reset ( 0 );
 		return pData;
 	}
+
+	/// swap
+	void SwapData ( CSphFixedVector<T> & rhs )
+	{
+		Swap ( m_pData, rhs.m_pData );
+		Swap ( m_iSize, rhs.m_iSize );
+	}
 };
 
 //////////////////////////////////////////////////////////////////////////
@@ -2708,6 +2715,7 @@ protected:
 	T &	m_tMutexRef;
 };
 
+
 /// scoped locked shared variable
 template < typename LOCK >
 class CSphScopedLockedShare : public CSphScopedLock<LOCK>
@@ -2795,6 +2803,31 @@ private:
 };
 
 
+/// scoped RW lock
+class CSphScopedRWLock : ISphNoncopyable
+{
+public:
+	/// lock on creation
+	CSphScopedRWLock ( CSphRwlock & tLock, bool bRead )
+		: m_tLock ( tLock )
+	{
+		if ( bRead )
+			m_tLock.ReadLock();
+		else
+			m_tLock.WriteLock();
+	}
+
+	/// unlock on going out of scope
+	~CSphScopedRWLock ()
+	{
+		m_tLock.Unlock ();
+	}
+
+protected:
+	CSphRwlock & m_tLock;
+};
+
+
 /// process-shared mutex that survives fork
 class CSphProcessSharedMutex
 {
@@ -2948,11 +2981,21 @@ public:
 		return m_pData;
 	}
 
+	DWORD * Begin ()
+	{
+		return m_pData;
+	}
+
 	int GetSize() const
 	{
 		return (m_iElements+31)/32;
 	}
 
+	int GetBits() const
+	{
+		return m_iElements;
+	}
+
 	int BitCount () const
 	{
 		int iBitSet = 0;
@@ -2984,7 +3027,7 @@ public:
 //////////////////////////////////////////////////////////////////////////
 // interlocked (atomic) operation
 
-#if (USE_WINDOWS) || (HAVE_SYNC_FETCH_AND_ADD)
+#if (USE_WINDOWS) || (HAVE_SYNC_FETCH)
 #define NO_ATOMIC 0
 #else
 #define NO_ATOMIC 1
@@ -3009,13 +3052,18 @@ public:
 	{
 		return m_uValue;
 	}
-#ifdef HAVE_SYNC_FETCH_AND_ADD
+#ifdef HAVE_SYNC_FETCH
 	inline UINT Inc()
 	{
 		return __sync_fetch_and_add ( &m_uValue, 1 );
 	}
+	inline UINT Dec()
+	{
+		return __sync_fetch_and_sub ( &m_uValue, 1 );
+	}
 #elif USE_WINDOWS
 	UINT Inc();
+	UINT Dec();
 #endif
 
 #if NO_ATOMIC
@@ -3027,6 +3075,14 @@ public:
 		m_tLock.Unlock();
 		return uTmp;
 	}
+	inline UINT Dec()
+	{
+		UINT uTmp;
+		m_tLock.Lock();
+		uTmp = m_uValue--;
+		m_tLock.Unlock();
+		return uTmp;
+	}
 #endif
 };