Skip to content

Commit

Permalink
-- Add mutex for user_connect handling
Browse files Browse the repository at this point in the history
  modified:   storage/connect/ha_connect.cc
  modified:   storage/connect/user_connect.cc
  modified:   storage/connect/plgdbutl.cpp
  modified:   storage/connect/user_connect.cc

-- Trace work storage allocation and freeing in DEVELOPMENT mode
  modified:   storage/connect/ha_connect.cc
  modified:   storage/connect/jsonudf.cpp
  modified:   storage/connect/plugutil.cpp
  • Loading branch information
Buggynours committed Jul 21, 2017
1 parent df09123 commit 8476d30
Show file tree
Hide file tree
Showing 5 changed files with 85 additions and 45 deletions.
71 changes: 49 additions & 22 deletions storage/connect/ha_connect.cc
Original file line number Diff line number Diff line change
Expand Up @@ -197,11 +197,14 @@ extern "C" {
char *ClassPath;
#endif // JDBC_SUPPORT

#if defined(__WIN__)
CRITICAL_SECTION parsec; // Used calling the Flex parser
#else // !__WIN__
pthread_mutex_t parmut = PTHREAD_MUTEX_INITIALIZER;
#endif // !__WIN__
//#if defined(__WIN__)
//CRITICAL_SECTION parsec; // Used calling the Flex parser
//#else // !__WIN__
//pthread_mutex_t parmut = PTHREAD_MUTEX_INITIALIZER;
//#endif // !__WIN__
pthread_mutex_t parmut;
pthread_mutex_t usrmut;
pthread_mutex_t tblmut;

/***********************************************************************/
/* Utility functions. */
Expand Down Expand Up @@ -679,10 +682,13 @@ static int connect_init_func(void *p)

#if defined(__WIN__)
sql_print_information("CONNECT: %s", compver);
InitializeCriticalSection((LPCRITICAL_SECTION)&parsec);
//InitializeCriticalSection((LPCRITICAL_SECTION)&parsec);
#else // !__WIN__
sql_print_information("CONNECT: %s", version);
#endif // !__WIN__
pthread_mutex_init(&parmut, NULL);
pthread_mutex_init(&usrmut, NULL);
pthread_mutex_init(&tblmut, NULL);

#if defined(LIBXML2_SUPPORT)
XmlInitParserLib();
Expand Down Expand Up @@ -738,20 +744,26 @@ static int connect_done_func(void *)
JAVAConn::ResetJVM();
#endif // JDBC_SUPPORT

pthread_mutex_destroy(&parmut);
pthread_mutex_destroy(&tblmut);
#if defined(__WIN__)
DeleteCriticalSection((LPCRITICAL_SECTION)&parsec);
//DeleteCriticalSection((LPCRITICAL_SECTION)&parsec);
#else // !__WIN__
PROFILE_End();
#endif // !__WIN__

for (pc= user_connect::to_users; pc; pc= pn) {
pthread_mutex_lock(&usrmut);
for (pc= user_connect::to_users; pc; pc= pn) {
if (pc->g)
PlugCleanup(pc->g, true);

pn= pc->next;
delete pc;
} // endfor pc

pthread_mutex_unlock(&usrmut);

pthread_mutex_destroy(&usrmut);
connect_hton= NULL;
DBUG_RETURN(error);
} // end of connect_done_func
Expand Down Expand Up @@ -864,6 +876,7 @@ ha_connect::~ha_connect(void)
static void PopUser(PCONNECT xp)
{
if (xp) {
pthread_mutex_lock(&usrmut);
xp->count--;

if (!xp->count) {
Expand All @@ -888,6 +901,7 @@ static void PopUser(PCONNECT xp)
delete xp;
} // endif count

pthread_mutex_unlock(&usrmut);
} // endif xp

} // end of PopUser
Expand All @@ -904,20 +918,29 @@ static PCONNECT GetUser(THD *thd, PCONNECT xp)
if (xp && thd == xp->thdp)
return xp;

for (xp= user_connect::to_users; xp; xp= xp->next)
pthread_mutex_lock(&usrmut);

for (xp= user_connect::to_users; xp; xp= xp->next)
if (thd == xp->thdp)
break;

if (!xp) {
xp= new user_connect(thd);
if (xp)
xp->count++;

if (xp->user_init()) {
delete xp;
xp= NULL;
} // endif user_init
pthread_mutex_unlock(&usrmut);

} else
xp->count++;
if (!xp) {
xp = new user_connect(thd);

if (xp->user_init()) {
delete xp;
xp = NULL;
} // endif user_init

} // endif xp

//} else
// xp->count++;

return xp;
} // end of GetUser
Expand Down Expand Up @@ -4373,7 +4396,11 @@ bool ha_connect::IsSameIndex(PIXDEF xp1, PIXDEF xp2)
MODE ha_connect::CheckMode(PGLOBAL g, THD *thd,
MODE newmode, bool *chk, bool *cras)
{
#if defined(DEVELOPMENT)
if (true) {
#else
if (trace) {
#endif
LEX_STRING *query_string= thd_query_string(thd);
htrc("%p check_mode: cmdtype=%d\n", this, thd_sql_command(thd));
htrc("Cmd=%.*s\n", (int) query_string->length, query_string->str);
Expand Down Expand Up @@ -5349,17 +5376,17 @@ static int connect_assisted_discovery(handlerton *, THD* thd,
PCOLRES crp;
PCONNECT xp= NULL;
PGLOBAL g= GetPlug(thd, xp);
PDBUSER dup= PlgGetUser(g);

if (!g)
return HA_ERR_INTERNAL_ERROR;

PDBUSER dup= PlgGetUser(g);
PCATLG cat= (dup) ? dup->Catalog : NULL;
PTOS topt= table_s->option_struct;
char buf[1024];
String sql(buf, sizeof(buf), system_charset_info);

sql.copy(STRING_WITH_LEN("CREATE TABLE whatever ("), system_charset_info);

if (!g)
return HA_ERR_INTERNAL_ERROR;

user= host= pwd= tbl= src= col= ocl= pic= fcl= skc= rnk= zfn= dsn= NULL;

// Get the useful create options
Expand Down
4 changes: 3 additions & 1 deletion storage/connect/jsonudf.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1478,8 +1478,10 @@ static my_bool CheckMemory(PGLOBAL g, UDF_INIT *initid, UDF_ARGS *args, uint n,
ml += g->More;

if (ml > g->Sarea_Size) {
#if !defined(DEVELOPMENT)
if (trace)
htrc("Freeing Sarea size=%d\n", g->Sarea_Size);
#endif
htrc("Freeing Sarea at %p size=%d\n", g->Sarea, g->Sarea_Size);

free(g->Sarea);

Expand Down
28 changes: 12 additions & 16 deletions storage/connect/plgdbutl.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -82,11 +82,11 @@ extern "C" {
extern char version[];
} // extern "C"

#if defined(__WIN__)
extern CRITICAL_SECTION parsec; // Used calling the Flex parser
#else // !__WIN__
//#if defined(__WIN__)
//extern CRITICAL_SECTION parsec; // Used calling the Flex parser
//#else // !__WIN__
extern pthread_mutex_t parmut;
#endif // !__WIN__
//#endif // !__WIN__

// The debug trace used by the main thread
FILE *pfile = NULL;
Expand Down Expand Up @@ -697,21 +697,17 @@ PDTP MakeDateFormat(PGLOBAL g, PCSZ dfmt, bool in, bool out, int flag)
/* Call the FLEX generated parser. In multi-threading mode the next */
/* instruction is included in an Enter/LeaveCriticalSection bracket. */
/*********************************************************************/
//#if defined(THREAD)
#if defined(__WIN__)
EnterCriticalSection((LPCRITICAL_SECTION)&parsec);
#else // !__WIN__
//#if defined(__WIN__)
// EnterCriticalSection((LPCRITICAL_SECTION)&parsec);
//#else // !__WIN__
pthread_mutex_lock(&parmut);
#endif // !__WIN__
//#endif // THREAD
//#endif // !__WIN__
rc = fmdflex(pdp);
//#if defined(THREAD)
#if defined(__WIN__)
LeaveCriticalSection((LPCRITICAL_SECTION)&parsec);
#else // !__WIN__
//#if defined(__WIN__)
// LeaveCriticalSection((LPCRITICAL_SECTION)&parsec);
//#else // !__WIN__
pthread_mutex_unlock(&parmut);
#endif // !__WIN__
//#endif // THREAD
//#endif // !__WIN__

if (trace)
htrc("Done: in=%s out=%s rc=%d\n", SVP(pdp->InFmt), SVP(pdp->OutFmt), rc);
Expand Down
10 changes: 8 additions & 2 deletions storage/connect/plugutil.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -184,8 +184,10 @@ int PlugExit(PGLOBAL g)
free(dup);

if (g->Sarea) {
if (trace)
htrc("Freeing Sarea size=%d\n", g->Sarea_Size);
#if !defined(DEVELOPMENT)
if (trace) {
#endif
htrc("Freeing Sarea at %p size=%d\n", g->Sarea, g->Sarea_Size);

free(g->Sarea);
} // endif Sarea
Expand Down Expand Up @@ -467,7 +469,11 @@ void *PlugAllocMem(PGLOBAL g, uint size)
if (!(areap = malloc(size)))
sprintf(g->Message, MSG(MALLOC_ERROR), "malloc");

#if defined(DEVELOPMENT)
if (true) {
#else
if (trace) {
#endif
if (areap)
htrc("Memory of %u allocated at %p\n", size, areap);
else
Expand Down
17 changes: 13 additions & 4 deletions storage/connect/user_connect.cc
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,8 @@
#include "user_connect.h"
#include "mycat.h"

extern pthread_mutex_t usrmut;

/****************************************************************************/
/* Initialize the user_connect static member. */
/****************************************************************************/
Expand Down Expand Up @@ -125,14 +127,18 @@ bool user_connect::user_init()
strcpy(ap->Ap_Name, "CONNECT");
g->Activityp= ap;
g->Activityp->Aptr= dup;

pthread_mutex_lock(&usrmut);
next= to_users;
to_users= this;

if (next)
next->previous= this;

last_query_id= thdp->query_id;
count= 1;
count = 1;
pthread_mutex_unlock(&usrmut);

last_query_id= thdp->query_id;
return false;
} // end of user_init

Expand All @@ -156,8 +162,11 @@ bool user_connect::CheckCleanup(bool force)

if (g->Sarea_Size != worksize) {
if (g->Sarea) {
if (trace)
htrc("CheckCleanup: Free Sarea %d\n", g->Sarea_Size);
#if !defined(DEVELOPMENT)
if (trace) {
#endif
htrc("CheckCleanup: Free Sarea at %p size=%d\n",
g->Sarea, g->Sarea_Size);

free(g->Sarea);
} // endif Size
Expand Down

0 comments on commit 8476d30

Please sign in to comment.