You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafodion.apache.org by se...@apache.org on 2016/11/17 05:37:05 UTC
[1/2] incubator-trafodion git commit: [TRAFODION-2342] hdfs access
threads init/cleanup changes
Repository: incubator-trafodion
Updated Branches:
refs/heads/master 362b57f9e -> f6598941c
[TRAFODION-2342] hdfs access threads init/cleanup changes
The worker threads to read hdfs directly was being started
unnecessarily in many cases. Also, ensured that these worker
threads are cleanly exited when the statement is deallocated.
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/commit/45e2c167
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/tree/45e2c167
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/diff/45e2c167
Branch: refs/heads/master
Commit: 45e2c1674922a364beb8c8acd1dba66587ba8142
Parents: 34772bc
Author: selvaganesang <se...@esgyn.com>
Authored: Fri Nov 11 16:29:46 2016 +0000
Committer: selvaganesang <se...@esgyn.com>
Committed: Tue Nov 15 14:52:15 2016 +0000
----------------------------------------------------------------------
core/sql/cli/Statement.cpp | 2 +-
core/sql/executor/ExExeUtilMisc.cpp | 6 ++++-
core/sql/executor/ExFastTransport.cpp | 6 ++---
core/sql/executor/ExHdfsScan.cpp | 19 +++++++++++-----
core/sql/executor/ExHdfsScan.h | 1 +
core/sql/exp/ExpLOB.cpp | 22 +++++++++---------
core/sql/exp/ExpLOBaccess.cpp | 36 +++++++++++++++++++++---------
core/sql/exp/ExpLOBaccess.h | 11 ++-------
core/sql/exp/ExpLOBinterface.cpp | 5 ++---
core/sql/exp/ExpLOBinterface.h | 2 +-
core/sql/optimizer/HDFSHook.cpp | 8 +++++--
11 files changed, 72 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/cli/Statement.cpp
----------------------------------------------------------------------
diff --git a/core/sql/cli/Statement.cpp b/core/sql/cli/Statement.cpp
index f7e9e2a..9c863e8 100644
--- a/core/sql/cli/Statement.cpp
+++ b/core/sql/cli/Statement.cpp
@@ -2151,7 +2151,7 @@ RETCODE Statement::doQuerySimilarityCheck(TrafQuerySimilarityInfo * qsi,
{
ExpLOBinterfaceInit
(lobGlob, &heap_, context_,
- TRUE, si->hdfsHostName(), si->hdfsPort());
+ FALSE, si->hdfsHostName(), si->hdfsPort());
lobGlobInitialized = TRUE;
}
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/executor/ExExeUtilMisc.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExExeUtilMisc.cpp b/core/sql/executor/ExExeUtilMisc.cpp
index 2d6e0c7..946a7b2 100644
--- a/core/sql/executor/ExExeUtilMisc.cpp
+++ b/core/sql/executor/ExExeUtilMisc.cpp
@@ -2270,6 +2270,10 @@ void ExExeUtilHiveTruncateTcb::freeResources()
hiveClient->init();
hiveClient->executeHiveSQL(hiveDropDDL);
}
+ if (lobGlob_) {
+ ExpLOBinterfaceCleanup(lobGlob_, getGlobals()->getDefaultHeap());
+ lobGlob_ = NULL;
+ }
}
Int32 ExExeUtilHiveTruncateTcb::fixup()
@@ -2278,7 +2282,7 @@ Int32 ExExeUtilHiveTruncateTcb::fixup()
ExpLOBinterfaceInit
(lobGlob_, getGlobals()->getDefaultHeap(),
- getGlobals()->castToExExeStmtGlobals()->getContext(),TRUE,
+ getGlobals()->castToExExeStmtGlobals()->getContext(),FALSE,
htTdb().getHdfsHost(),
htTdb().getHdfsPort());
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/executor/ExFastTransport.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExFastTransport.cpp b/core/sql/executor/ExFastTransport.cpp
index 2fe4a60..12ff6ee 100644
--- a/core/sql/executor/ExFastTransport.cpp
+++ b/core/sql/executor/ExFastTransport.cpp
@@ -527,15 +527,15 @@ ExHdfsFastExtractTcb::ExHdfsFastExtractTcb(
ExHdfsFastExtractTcb::~ExHdfsFastExtractTcb()
{
- if (lobGlob_ != NULL)
- {
+ if (lobGlob_) {
+ ExpLOBinterfaceCleanup(lobGlob_, getGlobals()->getDefaultHeap());
lobGlob_ = NULL;
}
if (sequenceFileWriter_ != NULL) {
NADELETE(sequenceFileWriter_, SequenceFileWriter, getHeap());
+ sequenceFileWriter_ = NULL;
}
-
} // ExHdfsFastExtractTcb::~ExHdfsFastExtractTcb()
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/executor/ExHdfsScan.cpp
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHdfsScan.cpp b/core/sql/executor/ExHdfsScan.cpp
index 8c630fd..8e43950 100644
--- a/core/sql/executor/ExHdfsScan.cpp
+++ b/core/sql/executor/ExHdfsScan.cpp
@@ -264,12 +264,12 @@ void ExHdfsScanTcb::freeResources()
}
if (runTimeRanges_)
deallocateRuntimeRanges();
-
- ExpLOBinterfaceCleanup
- (lobGlob_, getGlobals()->getDefaultHeap());
-
-
+ if (lobGlob_) {
+ ExpLOBinterfaceCleanup(lobGlob_, getGlobals()->getDefaultHeap());
+ lobGlob_ = NULL;
+ }
}
+
NABoolean ExHdfsScanTcb::needStatsEntry()
{
// stats are collected for ALL and OPERATOR options.
@@ -1967,6 +1967,15 @@ ExOrcScanTcb::~ExOrcScanTcb()
{
}
+Int32 ExOrcScanTcb::fixup()
+{
+ lobGlob_ = NULL;
+
+ return 0;
+}
+
+
+
short ExOrcScanTcb::extractAndTransformOrcSourceToSqlRow(
char * orcRow,
Int64 orcRowLen,
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/executor/ExHdfsScan.h
----------------------------------------------------------------------
diff --git a/core/sql/executor/ExHdfsScan.h b/core/sql/executor/ExHdfsScan.h
index aba7fe1..2df1ae0 100644
--- a/core/sql/executor/ExHdfsScan.h
+++ b/core/sql/executor/ExHdfsScan.h
@@ -329,6 +329,7 @@ protected:
, DONE
} step_;
+ virtual Int32 fixup();
/////////////////////////////////////////////////////
// Private methods.
/////////////////////////////////////////////////////
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/exp/ExpLOB.cpp
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOB.cpp b/core/sql/exp/ExpLOB.cpp
index cb6ded8..4c5dc3d 100644
--- a/core/sql/exp/ExpLOB.cpp
+++ b/core/sql/exp/ExpLOB.cpp
@@ -174,7 +174,8 @@ Lng32 ExpLOBoper::createLOB(void * exLobGlob, void *currContext, void * lobHeap,
exLobGlobL = exLobGlob;
rc = ExpLOBinterfaceCreate(exLobGlobL, lobName, lobLoc, Lob_HDFS_File,hdfsServer,lobMaxSize, hdfsPort);
-
+ if (exLobGlob == NULL)
+ ExpLOBinterfaceCleanup(exLobGlobL, lobHeap);
return rc;
}
void ExpLOBoper::calculateNewOffsets(ExLobInMemoryDescChunksEntry *dcArray, Lng32 numEntries)
@@ -231,6 +232,8 @@ Lng32 ExpLOBoper::compactLobDataFile(void *exLobGlob,ExLobInMemoryDescChunksEntr
rc = ExpLOBinterfacePerformGC(exLobGlobL,tgtLobName, (void *)dcArray, numEntries,hdfsServer,hdfsPort,lobLoc,lobMaxChunkMemSize);
+ if (exLobGlob == NULL)
+ ExpLOBinterfaceCleanup(exLobGlobL, lobHeap);
return rc;
}
@@ -246,11 +249,10 @@ Int32 ExpLOBoper::restoreLobDataFile(void *exLobGlob, char *lobName, void *lobHe
}
else
exLobGlobL = exLobGlob;
-
-
-
rc = ExpLOBinterfaceRestoreLobDataFile(exLobGlobL,hdfsServer,hdfsPort,lobLoc,lobName);
- return rc;
+ if (exLobGlob == NULL)
+ ExpLOBinterfaceCleanup(exLobGlobL, lobHeap);
+ return rc;
}
@@ -266,11 +268,9 @@ Int32 ExpLOBoper::purgeBackupLobDataFile(void *exLobGlob,char *lobName, void *cu
}
else
exLobGlobL = exLobGlob;
-
-
-
-
rc = ExpLOBinterfacePurgeBackupLobDataFile(exLobGlobL,(char *)hdfsServer,hdfsPort,lobLoc,lobName);
+ if (exLobGlob == NULL)
+ ExpLOBinterfaceCleanup(exLobGlobL, lobHeap);
return rc;
}
@@ -296,10 +296,10 @@ Lng32 ExpLOBoper::dropLOB(void * exLobGlob, void * lobHeap, void *currContext,
}
else
exLobGlobL = exLobGlob;
-
// Call ExeLOBinterface to drop the LOB
rc = ExpLOBinterfaceDrop(exLobGlobL,hdfsServer, hdfsPort, lobName, lobLoc);
-
+ if (exLobGlob == NULL)
+ ExpLOBinterfaceCleanup(exLobGlobL, lobHeap);
return rc;
}
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/exp/ExpLOBaccess.cpp
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOBaccess.cpp b/core/sql/exp/ExpLOBaccess.cpp
index e8e0f08..6143e02 100644
--- a/core/sql/exp/ExpLOBaccess.cpp
+++ b/core/sql/exp/ExpLOBaccess.cpp
@@ -60,7 +60,7 @@
#include "ExpLOBexternal.h"
#include "NAVersionedObject.h"
#include "ComQueue.h"
-
+#include "QRLogger.h"
#include "NAMemory.h"
#include <seabed/ms.h>
#include <seabed/fserr.h>
@@ -2280,8 +2280,10 @@ Ex_Lob_Error ExLobsOper (
return LOB_GLOB_PTR_ERROR;
}
}
-
- if ((globPtr != NULL) && (operation != Lob_Init))
+ if (globPtr != NULL)
+ {
+ lobGlobals = (ExLobGlobals *)globPtr;
+ if ((operation != Lob_Init) && (operation != Lob_Cleanup))
{
lobGlobals = (ExLobGlobals *)globPtr;
@@ -2314,6 +2316,7 @@ Ex_Lob_Error ExLobsOper (
}
lobPtr->lobTrace_ = lobGlobals->lobTrace_;
}
+ }
/*
// **Note** This is code that needs to get called before sneding a request to the
//mxlobsrvr process. It's inactive code currently
@@ -3014,9 +3017,9 @@ ExLobGlobals::ExLobGlobals() :
lobMap_(NULL),
fs_(NULL),
isCliInitialized_(FALSE),
- isHive_(FALSE),
threadTraceFile_(NULL),
lobTrace_(FALSE),
+ numWorkerThreads_(0),
heap_(NULL)
{
//initialize the log file
@@ -3043,12 +3046,21 @@ ExLobGlobals::~ExLobGlobals()
if (lobMap_)
delete lobMap_;
- for (int i=0; i<NUM_WORKER_THREADS; i++) {
- enqueueShutdownRequest();
- }
-
- for (int i=0; i<NUM_WORKER_THREADS; i++) {
- pthread_join(threadId_[i], NULL);
+ if (numWorkerThreads_ > 0) {
+ for (int i=0; numWorkerThreads_-i > 0 && i < NUM_WORKER_THREADS; i++) {
+ QRLogger::log(CAT_SQL_EXE, LL_DEBUG, 0, NULL,
+ "Worker Thread Shutdown Requested %ld ",
+ threadId_[i]);
+ enqueueShutdownRequest();
+ }
+
+ for (int i=0; numWorkerThreads_ > 0 && i < NUM_WORKER_THREADS; i++) {
+ pthread_join(threadId_[i], NULL);
+ QRLogger::log(CAT_SQL_EXE, LL_DEBUG, 0, NULL,
+ "Worker Thread Completed %ld ",
+ threadId_[i]);
+ numWorkerThreads_--;
+ }
}
// Free the post fetch bugf list AFTER the worker threads have left to
// avoid slow worker thread being stuck and master deallocating these
@@ -3109,6 +3121,10 @@ Ex_Lob_Error ExLobGlobals::startWorkerThreads()
rc = pthread_create(&threadId_[i], NULL, workerThreadMain, this);
if (rc != 0)
return LOB_HDFS_THREAD_CREATE_ERROR;
+ QRLogger::log(CAT_SQL_EXE, LL_DEBUG, 0, NULL,
+ "Worker Thread Created %ld ",
+ threadId_[i]);
+ numWorkerThreads_++;
}
return LOB_OPER_OK;
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/exp/ExpLOBaccess.h
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOBaccess.h b/core/sql/exp/ExpLOBaccess.h
index 7b0132e..a8e0ea7 100644
--- a/core/sql/exp/ExpLOBaccess.h
+++ b/core/sql/exp/ExpLOBaccess.h
@@ -618,15 +618,7 @@ class ExLobGlobals
{
isCliInitialized_ = TRUE;
}
- NABoolean isHive()
- {
- return isHive_;
- }
-
- void setIsHive(NABoolean TorF)
- {
- isHive_ = TorF;
- }
+
void setHeap(void * heap)
{
heap_ = (NAHeap *) heap;
@@ -654,6 +646,7 @@ class ExLobGlobals
FILE *threadTraceFile_;
NAHeap *heap_;
NABoolean lobTrace_;
+ long numWorkerThreads_;
};
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/exp/ExpLOBinterface.cpp
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOBinterface.cpp b/core/sql/exp/ExpLOBinterface.cpp
index 4c29194..d8cb3bd 100644
--- a/core/sql/exp/ExpLOBinterface.cpp
+++ b/core/sql/exp/ExpLOBinterface.cpp
@@ -35,7 +35,7 @@ using std::ofstream;
#include "ex_globals.h"
Lng32 ExpLOBinterfaceInit(void *& exLobGlob, void * lobHeap,
- void *currContext,NABoolean isHive,
+ void *currContext,NABoolean isHiveRead,
char *hdfsServer,
Int32 port)
{
@@ -64,9 +64,8 @@ Lng32 ExpLOBinterfaceInit(void *& exLobGlob, void * lobHeap,
0);
if (exLobGlob)
{
- ((ExLobGlobals *)exLobGlob)->setIsHive(isHive);
NAHeap *heap = new ((NAHeap *)lobHeap) NAHeap("LOB Heap", (NAHeap *)lobHeap);
- if (isHive)
+ if (isHiveRead)
((ExLobGlobals *)exLobGlob)->startWorkerThreads();
heap->setThreadSafe();
((ExLobGlobals *)exLobGlob)->setHeap(heap);
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/exp/ExpLOBinterface.h
----------------------------------------------------------------------
diff --git a/core/sql/exp/ExpLOBinterface.h b/core/sql/exp/ExpLOBinterface.h
index 50efa36..33d63b4 100644
--- a/core/sql/exp/ExpLOBinterface.h
+++ b/core/sql/exp/ExpLOBinterface.h
@@ -77,7 +77,7 @@ enum ExpLOBinterfaceInputFlags
ERROR_IF_TGT_FILE_EXISTS_ = 0x0004
};
-Lng32 ExpLOBinterfaceInit(void *& lobGlob, void * lobHeap, void *currContext,NABoolean isHive=FALSE, char *hdfsServer="default", Int32 port=0);
+Lng32 ExpLOBinterfaceInit(void *& lobGlob, void * lobHeap, void *currContext,NABoolean isHiveRead, char *hdfsServer="default", Int32 port=0);
Lng32 ExpLOBinterfaceCleanup(void *& lobGlob, void * lobHeap);
http://git-wip-us.apache.org/repos/asf/incubator-trafodion/blob/45e2c167/core/sql/optimizer/HDFSHook.cpp
----------------------------------------------------------------------
diff --git a/core/sql/optimizer/HDFSHook.cpp b/core/sql/optimizer/HDFSHook.cpp
index 4b8ce88..1a1cac4 100644
--- a/core/sql/optimizer/HDFSHook.cpp
+++ b/core/sql/optimizer/HDFSHook.cpp
@@ -854,11 +854,13 @@ NABoolean HHDFSTableStats::populate(struct hive_tbl_desc *htd)
hdfsPort,
tableDir,
diags_,
- hdfsPortOverride_))
+ hdfsPortOverride_)) {
return FALSE;
+ }
- if (! connectHDFS(hdfsHost, hdfsPort))
+ if (! connectHDFS(hdfsHost, hdfsPort)) {
return FALSE; // diags_ is set
+ }
// put back fully qualified URI
tableDir = hsd->location_;
@@ -1083,6 +1085,7 @@ void HHDFSTableStats::print(FILE *ofd)
fprintf(ofd,"====================================================================\n");
}
+
NABoolean HHDFSTableStats::connectHDFS(const NAString &host, Int32 port)
{
NABoolean result = TRUE;
@@ -1114,3 +1117,4 @@ void HHDFSTableStats::disconnectHDFS()
// No op. The disconnect happens at the context level wehn the session
// is dropped or the thread exits.
}
+
[2/2] incubator-trafodion git commit: Merge [TRAFODION-2342] PR 835
hdfs access threads init/cleanup changes
Posted by se...@apache.org.
Merge [TRAFODION-2342] PR 835 hdfs access threads init/cleanup changes
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/commit/f6598941
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/tree/f6598941
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafodion/diff/f6598941
Branch: refs/heads/master
Commit: f6598941c01be75e6b929645ddda2d0a57100dea
Parents: 362b57f 45e2c16
Author: selvaganesang <se...@apache.org>
Authored: Thu Nov 17 05:36:34 2016 +0000
Committer: selvaganesang <se...@apache.org>
Committed: Thu Nov 17 05:36:34 2016 +0000
----------------------------------------------------------------------
core/sql/cli/Statement.cpp | 2 +-
core/sql/executor/ExExeUtilMisc.cpp | 6 ++++-
core/sql/executor/ExFastTransport.cpp | 6 ++---
core/sql/executor/ExHdfsScan.cpp | 19 +++++++++++-----
core/sql/executor/ExHdfsScan.h | 1 +
core/sql/exp/ExpLOB.cpp | 22 +++++++++---------
core/sql/exp/ExpLOBaccess.cpp | 36 +++++++++++++++++++++---------
core/sql/exp/ExpLOBaccess.h | 11 ++-------
core/sql/exp/ExpLOBinterface.cpp | 5 ++---
core/sql/exp/ExpLOBinterface.h | 2 +-
core/sql/optimizer/HDFSHook.cpp | 8 +++++--
11 files changed, 72 insertions(+), 46 deletions(-)
----------------------------------------------------------------------