tracing with perfetto

Notice, perfetto itself is not included, and tracing disabled.
For now you need to download it explicitly.

Look at src/perfetto/README.txt for instructions

src/CMakeLists.txt

@@ -60,7 +60,7 @@ add_library ( lmanticore STATIC sphinx.cpp sphinxexcerpt.cpp sphinxquery.cpp sph
 		timeout_queue.cpp dynamic_idx.cpp columnarrt.cpp columnarmisc.cpp exprtraits.cpp columnarexpr.cpp
 		sphinx_alter.cpp columnarsort.cpp binlog.cpp chunksearchctx.cpp client_task_info.cpp
 		indexfiles.cpp attrindex_builder.cpp queryfilter.cpp aggregate.cpp secondarylib.cpp costestimate.cpp
-		docidlookup.cpp )
+		docidlookup.cpp tracer.cpp )
 
 add_library ( lstem STATIC sphinxsoundex.cpp sphinxmetaphone.cpp sphinxstemen.cpp sphinxstemru.cpp sphinxstemru.inl
 		sphinxstemcz.cpp sphinxstemar.cpp )
@@ -149,7 +149,7 @@ set ( HEADERS sphinxexcerpt.h sphinxfilter.h sphinxint.h sphinxjsonquery.h sphin
 		indexsettings.h columnarlib.h fileio.h memio.h memio_impl.h queryprofile.h columnarfilter.h columnargrouper.h fileutils.h
 		libutils.h conversion.h columnarsort.h sortcomp.h binlog_defs.h binlog.h ${MANTICORE_BINARY_DIR}/config/config.h
 		chunksearchctx.h indexfiles.h attrindex_builder.h queryfilter.h aggregate.h secondarylib.h
-		costestimate.h docidlookup.h )
+		costestimate.h docidlookup.h tracer.h )
 
 set ( SEARCHD_H searchdaemon.h searchdconfig.h searchdddl.h searchdexpr.h searchdha.h searchdreplication.h searchdsql.h
 		searchdtask.h client_task_info.h taskflushattrs.h taskflushbinlog.h taskflushmutable.h taskglobalidf.h
@@ -250,6 +250,9 @@ if (WITH_ZLIB OR WITH_ZSTD)
 	endif ()
 endif ()
 
+add_subdirectory ( perfetto )
+target_link_libraries ( lextra INTERFACE perfetto )
+
 # stub library for all except daemon - to avoid link with openssl
 add_library ( stub_ssl searchdssl.cpp )
 target_link_libraries ( stub_ssl PRIVATE lextra )

src/accumulator.cpp

@@ -14,6 +14,7 @@
 #include "sphinxrt.h"
 #include "columnarmisc.h"
 #include "memio.h"
+#include "tracer.h"
 
 #include <memory>
 
@@ -72,6 +73,7 @@ int RtAccum_t::GetPackedLen() const
 
 void RtAccum_t::Sort()
 {
+	TRACE_CONN ( "conn", "RtAccum_t::Sort" );
 	if ( !m_bKeywordDict )
 		m_dAccum.Sort ( Lesser ( [] ( const CSphWordHit& a, const CSphWordHit& b )
 		{
@@ -326,6 +328,8 @@ struct AccumDocHits_t
 
 void RtAccum_t::CleanupDuplicates ( int iRowSize )
 {
+	TRACE_CONN ( "conn", "RtAccum_t::CleanupDuplicates" );
+
 	if ( m_uAccumDocs <= 1 )
 		return;
 

src/gtests/gtests_globalstate.cpp

@@ -14,6 +14,7 @@
 
 #include "sphinxint.h"
 #include "threadutils.h"
+#include "tracer.h"
 
 // global stuff
 
@@ -59,6 +60,7 @@ class Environment : public ::testing::Environment
 		char cTopOfMainStack;
 		Threads::Init ();
 		Threads::PrepareMainThread ( &cTopOfMainStack );
+		Tracer::Init();
 		CreateSynonymsFile ();
 		CreateSynonymsFile ( g_sMagic );
 		auto iThreads = sphCpuThreadsCount();

src/netreceive_http.cpp

@@ -13,6 +13,7 @@
 #include "netreceive_http.h"
 #include "searchdssl.h"
 #include "searchdhttp.h"
+#include "tracer.h"
 
 extern int g_iClientTimeoutS; // from searchd.cpp
 extern volatile bool g_bMaintenance;
@@ -71,6 +72,7 @@ void HttpServe ( std::unique_ptr<AsyncNetBuffer_c> pBuf )
 
 	HttpRequestParser_c tParser;
 	CSphVector<BYTE> dResult;
+	TRACE_CONN ( "conn", "HttpServe" );
 
 	auto HttpReply = [&dResult, &tOut] ( ESphHttpStatus eCode, Str_t sMsg )
 	{
@@ -136,6 +138,7 @@ void HttpServe ( std::unique_ptr<AsyncNetBuffer_c> pBuf )
 			sphLogDebug ("100 Continue sent");
 		}
 
+//		tracer.Instant ( [&tIn](StringBuilder_c& sOut) {sOut<< ",\"args\":{\"step\":"<<tIn.HasBytes()<<"}";} );
 		tParser.ProcessClientHttp ( tIn, dResult );
 
 		tOut.SwapData (dResult);

src/networking_daemon.cpp

@@ -15,6 +15,7 @@
 #include "net_action_accept.h"
 #include "netstate_api.h"
 #include "coroutine.h"
+#include "tracer.h"
 
 #if _WIN32
 // Win-specific headers and calls
@@ -540,6 +541,7 @@ int SockWrapper_c::Impl_c::SockPollNetloop ( int64_t tmTimeUntilUs, bool bWrite
 // as usual sphPoll - returns 1 on success, 0 on timeout, -1 on error.
 int SockWrapper_c::Impl_c::SockPoll ( int64_t tmTimeUntilUs, bool bWrite )
 {
+	TRACE_CONN ( "conn", "SockPoll" );
 	session::SetTaskState ( TaskState_e::NET_IDLE );
 	auto _ = AtScopeExit([bWrite] { session::SetTaskState ( bWrite ? TaskState_e::NET_WRITE : TaskState_e::NET_READ ); });
 	return m_pNetLoop ? SockPollNetloop ( tmTimeUntilUs, bWrite ) : SockPollClassic ( tmTimeUntilUs, bWrite );

src/perfetto/CMakeLists.txt

@@ -0,0 +1,18 @@
+cmake_minimum_required ( VERSION 3.20 )
+
+# Provide valid perfetto library despite presence of the sources
+if (EXISTS perfetto.cc AND NOT CMAKE_SYSTEM_NAME STREQUAL FreeBSD)
+	# Define a static library for Perfetto.
+	add_library ( perfetto STATIC perfetto.cc )
+	target_include_directories ( perfetto INTERFACE "${MANTICORE_CURRENT_SOURCE_DIR}" )
+	target_compile_definitions ( perfetto PUBLIC PERFETTO )
+
+	if (WIN32)
+		target_compile_definitions ( perfetto PRIVATE NOMINMAX=1 WIN32_LEAN_AND_MEAN=1 PUBLIC _WIN32 )
+		target_compile_options ( perfetto PRIVATE $<$<CXX_COMPILER_ID:MSVC>:/bigobj> )
+		target_link_libraries ( perfetto PRIVATE WS2_32 )
+	endif ()
+else()
+	# Define an empty (stub) perfetto.
+	add_library ( perfetto INTERFACE )
+endif()

src/perfetto/README.txt

@@ -0,0 +1,10 @@
+Google Perfetto tracing SDK
+Apache v2 license
+
+How to retrieve:
+
+1. `git clone https://android.googlesource.com/platform/external/perfetto -b v27.1`
+2. Copy `perfetto.cc` and `perfetto.h` from folder `sdk` into this folder
+3. Reconfigure
+
+Generated traces can be opened with https://ui.perfetto.dev

src/searchd.cpp

@@ -49,6 +49,7 @@
 #include "config_reloader.h"
 #include "secondarylib.h"
 #include "task_dispatcher.h"
+#include "tracer.h"
 
 // services
 #include "taskping.h"
@@ -10782,6 +10783,7 @@ void sphHandleMysqlCommitRollback ( StmtErrorReporter_i& tOut, Str_t sQuery, boo
 	auto& tAcc = pSession->m_tAcc;
 	auto& sError = pSession->m_sError;
 	auto& tCrashQuery = GlobalCrashQueryGetRef();
+	TRACE_CONN ( "conn", "sphHandleMysqlCommitRollback" );
 
 	MEMORY ( MEM_SQL_COMMIT );
 	pSession->m_bInTransaction = false;
@@ -14245,6 +14247,27 @@ void HandleWaitStatus ( RowBuffer_i& tOutBuf, const DebugCmd::DebugCommand_t& tC
 }
 #endif
 
+void HandleTrace ( RowBuffer_i& tOut, const DebugCmd::DebugCommand_t& tCmd )
+{
+	tOut.HeadTuplet ( "command", "result" );
+#ifdef PERFETTO
+	if ( tCmd.m_sParam.IsEmpty() )
+	{
+		if ( !tCmd.m_iPar1 )
+		{
+			Tracer::Stop();
+		}
+	} else
+	{
+		Tracer::Start ( tCmd.m_sParam, tCmd.m_iPar1 );
+	}
+	tOut.DataTuplet ( "debug trace ...", "SUCCESS" );
+#else
+	tOut.DataTuplet ( "debug trace ...", "FAIL, need to rebuild with Perfetto, look to src/perfetto/README.txt" );
+#endif
+	tOut.Eof();
+}
+
 void HandleToken ( RowBuffer_i & tOut, const CSphString & sParam )
 {
 	auto sSha = strSHA1 ( sParam );
@@ -14380,6 +14403,7 @@ void HandleMysqlDebug ( RowBuffer_i &tOut, Str_t sCommand, const QueryProfile_c
 	case Cmd_e::WAIT: HandleWait ( tOut, tCmd ); return;
 	case Cmd_e::WAIT_STATUS: HandleWaitStatus ( tOut, tCmd ); return;
 #endif
+	case Cmd_e::TRACE: HandleTrace ( tOut, tCmd );	return;
 	default: break;
 	}
 
@@ -19344,6 +19368,8 @@ int WINAPI ServiceMain ( int argc, char **argv ) EXCLUDES (MainThread)
 
 	tzset();
 
+	Tracer::Init();
+
 	CSphString sError;
 	// initialize it before other code to fetch version string for banner
 	bool bColumnarError = !InitColumnar ( sError );

src/searchdhttp.cpp

@@ -22,6 +22,7 @@
 #include "accumulator.h"
 #include "networking_daemon.h"
 #include "client_session.h"
+#include "tracer.h"
 
 #define LOG_COMPONENT_HTTP ""
 #define LOG_LEVEL_HTTP false
@@ -1044,6 +1045,7 @@ class HttpSearchHandler_c : public HttpHandler_c
 public:
 	bool Process () final
 	{
+		TRACE_CONN ( "conn", "HttpSearchHandler_c::Process" );
 		CSphString sWarning;
 		std::unique_ptr<QueryParser_i> pQueryParser = PreParseQuery();
 		if ( !pQueryParser )
@@ -1390,6 +1392,7 @@ class HttpRawSqlHandler_c final: public HttpHandler_c
 
 	bool Process () final
 	{
+		TRACE_CONN ( "conn", "HttpRawSqlHandler_c::Process" );
 		if ( IsEmpty ( m_sQuery ) )
 		{
 			ReportError ( "query missing", SPH_HTTP_STATUS_400 );
@@ -1526,6 +1529,7 @@ class HttpHandler_JsonInsert_c final : public HttpHandler_c
 
 	bool Process () final
 	{
+		TRACE_CONN ( "conn", "HttpHandler_JsonInsert_c::Process" );
 		SqlStmt_t tStmt;
 		DocID_t tDocId = 0;
 		CSphString sError;
@@ -1587,6 +1591,7 @@ class HttpHandler_JsonUpdate_c : public HttpHandler_c, HttpJsonUpdateTraits_c
 
 	bool Process () final
 	{
+		TRACE_CONN ( "conn", "HttpHandler_JsonUpdate_c::Process" );
 		SqlStmt_t tStmt;
 		tStmt.m_bJson = true;
 		tStmt.m_sEndpoint = HttpEndpointToStr ( SPH_HTTP_ENDPOINT_JSON_UPDATE );
@@ -1737,6 +1742,7 @@ class HttpHandler_JsonBulk_c final : public HttpHandler_c, public HttpJsonUpdate
 
 	bool Process () final
 	{
+		TRACE_CONN ( "conn", "HttpHandler_JsonBulk_c::Process" );
 		const char* szError = IsNDJson ();
 		if ( szError )
 		{
@@ -1968,6 +1974,8 @@ static std::unique_ptr<HttpHandler_c> CreateHttpHandler ( ESphHttpEndpoint eEndp
 
 static bool ProcessHttpQuery ( ESphHttpEndpoint eEndpoint, CharStream_c& tSource, const OptionsHash_t& tOptions, CSphVector<BYTE> & dResult, bool bNeedHttpResponse, http_method eRequestType, const CSphString& sInvalidEndpoint )
 {
+	TRACE_CONN ( "conn", "ProcessHttpQuery" );
+
 	std::unique_ptr<HttpHandler_c> pHandler = CreateHttpHandler ( eEndpoint, tSource, tOptions, eRequestType );
 	if ( !pHandler )
 	{
@@ -2360,6 +2368,7 @@ bool HttpHandlerPQ_c::Delete ( const CSphString & sIndex, const JsonObj_c & tRoo
 
 bool HttpHandlerPQ_c::Process()
 {
+	TRACE_CONN ( "conn", "HttpHandlerPQ_c::Process" );
 	CSphString * sEndpoint = m_tOptions ( "endpoint" );
 	if ( !sEndpoint || sEndpoint->IsEmpty() )
 	{

src/searchdreplication.cpp

@@ -19,6 +19,7 @@
 #include <math.h>
 
 #include "digest_sha1.h"
+#include "tracer.h"
 
 #if !HAVE_WSREP
 #include "replication/wsrep_api_stub.h"
@@ -1194,6 +1195,7 @@ DEFINE_RENDER ( RPLRep_t )
 // replicate serialized data into cluster and call commit monitor along
 static bool Replicate ( int iKeysCount, const wsrep_key_t * pKeys, const wsrep_buf_t & tQueries, wsrep_t * pProvider, CommitMonitor_c & tMonitor, bool bUpdate, CSphString & sError )
 {
+	TRACE_CONN ( "conn", "Replicate" );
 	assert ( pProvider );
 
 	// just displays 'RPL' flag.
@@ -1219,6 +1221,7 @@ static bool Replicate ( int iKeysCount, const wsrep_key_t * pKeys, const wsrep_b
 
 	if ( bOk )
 	{
+		TRACE_CONN ( "conn", "pProvider->replicate" );
 		tRes = pProvider->replicate ( pProvider, iConnId, &tHnd, WSREP_FLAG_COMMIT, &tLogMeta );
 		bOk = CheckResult ( tRes, tLogMeta, "replicate", sError );
 	}
@@ -1676,6 +1679,7 @@ bool ParseCmdReplicated ( const BYTE * pData, int iLen, bool bIsolated, const CS
 // callback for Galera commit_cb to commit replicated command
 bool HandleCmdReplicated ( RtAccum_t & tAcc )
 {
+	TRACE_SCHED ( "conn", "HandleCmdReplicated" );
 	if ( tAcc.m_dCmd.IsEmpty() )
 	{
 		sphWarning ( "empty accumulator" );
@@ -1765,6 +1769,7 @@ bool HandleCmdReplicated ( RtAccum_t & tAcc )
 // single point there all commands passed these might be replicated, even if no cluster
 static bool HandleCmdReplicate ( RtAccum_t & tAcc, CSphString & sError, int * pDeletedCount, CSphString * pWarning, int * pUpdated, ServedClone_c * pDesc ) EXCLUDES ( g_tClustersLock )
 {
+	TRACE_CONN ( "conn", "HandleCmdReplicate" );
 	CommitMonitor_c tMonitor ( tAcc, pDeletedCount, pWarning, pUpdated );
 
 	// without cluster path
@@ -1835,6 +1840,7 @@ static bool HandleCmdReplicate ( RtAccum_t & tAcc, CSphString & sError, int * pD
 //	dBufKeys.ZeroVec();
 	int iKey = 0;
 
+	BEGIN_CONN ( "conn", "HandleCmdReplicate.serialize", "commands", tAcc.m_dCmd.GetLength() );
 	ARRAY_FOREACH ( i, tAcc.m_dCmd )
 	{
 		const ReplicationCommand_t & tCmd = *tAcc.m_dCmd[i];
@@ -1935,7 +1941,9 @@ static bool HandleCmdReplicate ( RtAccum_t & tAcc, CSphString & sError, int * pD
 		int iReqLen = dBufQueries.GetLength() - iLenOff;
 		memcpy ( dBufQueries.Begin() + iLenOff - sizeof ( iReqLen ), &iReqLen, sizeof ( iReqLen ) );
 	}
+	END_CONN ( "conn" );
 
+	BEGIN_CONN ( "conn", "HandleCmdReplicate.make_keys", "keys", iKeysCount );
 	// set keys wsrep_buf_t ptr and len
 	for ( int i=0; i<iKeysCount; i++ )
 	{
@@ -1945,12 +1953,15 @@ static bool HandleCmdReplicate ( RtAccum_t & tAcc, CSphString & sError, int * pD
 		dKeys[i].key_parts = dBufProxy.Begin() + i;
 		dKeys[i].key_parts_num = 1;
 	}
+	END_CONN ( "conn" );
 
 	wsrep_buf_t tQueries;
 	tQueries.ptr = dBufQueries.Begin();
 	tQueries.len = dBufQueries.GetLength();
 
+	BEGIN_CONN ( "conn", "HandleCmdReplicate.cluster_lock" );
 	Threads::ScopedCoroMutex_t tClusterLock { pCluster->m_tReplicationMutex };
+	END_CONN ( "conn" );
 
 	if ( !bTOI )
 		return Replicate ( iKeysCount, dKeys.Begin(), tQueries, pCluster->m_pProvider, tMonitor, bUpdate, sError );
@@ -1976,6 +1987,7 @@ bool HandleCmdReplicate ( RtAccum_t & tAcc, CSphString & sError, CSphString & sW
 // commit for common commands
 bool CommitMonitor_c::Commit ( CSphString& sError )
 {
+	TRACE_CONN ( "conn", "CommitMonitor_c::Commit" );
 	RtIndex_i* pIndex = m_tAcc.GetIndex ();
 
 	// short path for usual accum without commands

src/sphinxql_debug.cpp

@@ -220,4 +220,5 @@ CmdNotice_t DebugCmd::dCommands[(BYTE) Cmd_e::INVALID_CMD] = {
 	{ NO_WIN, "debug wait <cluster> [like 'xx'] [option timeout=3]", "wait <cluster> ready, but no more than 3 secs." },
 	{ NO_WIN, "debug wait <cluster> status <N> [like 'xx'] [option timeout=13]", "wait <cluster> commit achieve <N>, but no more than 13 secs" },
 	{ NONE, "debug meta", "Show max_matches/pseudo_shards. Needs set profiling=1" },
+	{ NONE, "debug trace OFF|'path/to/file' [<N>]", "trace flow to file until N bytes written, or 'trace OFF'" },
 };

src/sphinxql_debug.h

@@ -46,6 +46,7 @@ enum class Cmd_e : BYTE
 	WAIT,
 	WAIT_STATUS,
 	META,
+	TRACE,
 
 	INVALID_CMD
 };
@@ -55,7 +56,7 @@ struct DebugCommand_t
 	Cmd_e	m_eCommand {Cmd_e::INVALID_CMD};
 	CSphString m_sParam;
 	CSphString m_sParam2;
-	int64_t m_iPar1;
+	int64_t m_iPar1 = -1;
 	int64_t m_iPar2;
 	const char * m_szStmt;
 	SmallStringHash_T<Option_t> m_hOptions;

src/sphinxql_debug.l

@@ -65,6 +65,7 @@ FLOAT_CONSTANT      ({I}\.{I}?|{I}?\.{I}){EXPONENT}?
 "COMPRESS"			{ STORE_BOUNDS; return TOK_COMPRESS; }
 "SPLIT"				{ STORE_BOUNDS; return TOK_SPLIT; }
 "META"				{ STORE_BOUNDS; return TOK_META; }
+"TRACE"				{ STORE_BOUNDS; return TOK_TRACE; }
 
 "ON"				{ STORE_BOUNDS; return TOK_ON; }
 "OFF"				{ STORE_BOUNDS; return TOK_OFF; }