Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

SNOW-894815: Fix race conditions and enable TestConcurrentReadOnParams #909

Closed
wants to merge 7 commits into from
Closed

SNOW-894815: Fix race conditions and enable TestConcurrentReadOnParams #909

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
e0b78c5
enable TestConcurrentReadOnParams
sfc-gh-ext-simba-jl Sep 7, 2023
d5d299b
add mutext lock on params map
sfc-gh-ext-simba-jl Sep 7, 2023
f659625
add mutext lock on params map
sfc-gh-ext-simba-jl Sep 7, 2023
19c3b0a
remove mutex lock around getCurrentLocation
sfc-gh-ext-simba-jl Sep 26, 2023
52d34dd
add mutex lock to snowflakeConn config
sfc-gh-ext-simba-jl Sep 28, 2023
02f13d1
remove mutex lock in getQueryContextCacheSize
sfc-gh-ext-simba-jl Oct 4, 2023
644b4c0
add mutex lock in getQueryContextCacheSize
sfc-gh-ext-simba-jl Oct 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions chunk_downloader.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,7 @@ func (scd *snowflakeChunkDownloader) start() error {
// if the rowsetbase64 retrieved from the server is empty, move on to downloading chunks
var err error
var loc *time.Location
if scd.sc != nil && scd.sc.cfg != nil {
if scd.sc != nil && scd.sc.getConnConfig() != nil {
loc = getCurrentLocation(scd.sc.cfg.Params)
}
firstArrowChunk := buildFirstArrowChunk(scd.RowSet.RowSetBase64, loc, scd.pool)
Expand Down Expand Up @@ -271,7 +271,7 @@ func (scd *snowflakeChunkDownloader) startArrowBatches() error {
var err error
chunkMetaLen := len(scd.ChunkMetas)
var loc *time.Location
if scd.sc != nil && scd.sc.cfg != nil {
if scd.sc != nil && scd.sc.getConnConfig() != nil {
loc = getCurrentLocation(scd.sc.cfg.Params)
}
firstArrowChunk := buildFirstArrowChunk(scd.RowSet.RowSetBase64, loc, scd.pool)
Expand Down Expand Up @@ -432,7 +432,7 @@ func decodeChunk(scd *snowflakeChunkDownloader, idx int, bufStream *bufio.Reader
return err
}
var loc *time.Location
if scd.sc != nil && scd.sc.cfg != nil {
if scd.sc != nil && scd.sc.getConnConfig() != nil {
loc = getCurrentLocation(scd.sc.cfg.Params)
}
arc := arrowResultChunk{
Expand Down
4 changes: 3 additions & 1 deletion connection.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -249,7 +249,9 @@ func (sc *snowflakeConn) cleanup() {
sc.rest.Client.CloseIdleConnections()
}
sc.rest = nil
paramsMutex.Lock()
sc.cfg = nil
paramsMutex.Unlock()
}

func (sc *snowflakeConn) Close() (err error) {
Expand All @@ -258,7 +260,7 @@ func (sc *snowflakeConn) Close() (err error) {
sc.stopHeartBeat()
defer sc.cleanup()

if sc.cfg != nil && !sc.cfg.KeepSessionAlive {
if sc.getConnConfig() != nil && !sc.cfg.KeepSessionAlive {
if err = sc.rest.FuncCloseSession(sc.ctx, sc.rest, sc.rest.RequestTimeout); err != nil {
logger.Error(err)
}
Expand Down
1 change: 0 additions & 1 deletion connection_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -476,7 +476,6 @@ func TestExecWithServerSideError(t *testing.T) {
}

func TestConcurrentReadOnParams(t *testing.T) {
t.Skip("Fails randomly")
config, err := ParseDSN(dsn)
if err != nil {
t.Fatal("Failed to parse dsn")
Expand Down
10 changes: 8 additions & 2 deletions connection_util.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,12 @@ import (
"time"
)

func (sc *snowflakeConn) getConnConfig() *Config {
paramsMutex.Lock()
defer paramsMutex.Unlock()
return sc.cfg
}

func (sc *snowflakeConn) isClientSessionKeepAliveEnabled() bool {
paramsMutex.Lock()
v, ok := sc.cfg.Params[sessionClientSessionKeepAlive]
Expand All @@ -24,7 +30,7 @@ func (sc *snowflakeConn) isClientSessionKeepAliveEnabled() bool {
}

func (sc *snowflakeConn) startHeartBeat() {
if sc.cfg != nil && !sc.isClientSessionKeepAliveEnabled() {
if sc.getConnConfig() != nil && !sc.isClientSessionKeepAliveEnabled() {
return
}
if sc.rest != nil {
Expand All @@ -36,7 +42,7 @@ func (sc *snowflakeConn) startHeartBeat() {
}

func (sc *snowflakeConn) stopHeartBeat() {
if sc.cfg != nil && !sc.isClientSessionKeepAliveEnabled() {
if sc.getConnConfig() != nil && !sc.isClientSessionKeepAliveEnabled() {
return
}
if sc.rest != nil && sc.rest.HeartBeat != nil {
Expand Down
6 changes: 5 additions & 1 deletion htap.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,11 @@ func (qcc *queryContextCache) prune(size int) {
}

func (qcc *queryContextCache) getQueryContextCacheSize(sc *snowflakeConn) int {
if sizeStr, ok := sc.cfg.Params[queryContextCacheSizeParamName]; ok {
paramsMutex.Lock()
sizeStr, ok := sc.cfg.Params[queryContextCacheSizeParamName]
paramsMutex.Unlock()

if ok {
size, err := strconv.Atoi(*sizeStr)
if err != nil {
logger.Warnf("cannot parse %v as int as query context cache size: %v", sizeStr, err)
Expand Down
2 changes: 1 addition & 1 deletion rows.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ type snowflakeRows struct {
}

func (rows *snowflakeRows) getLocation() *time.Location {
if rows.location == nil && rows.sc != nil && rows.sc.cfg != nil {
if rows.location == nil && rows.sc != nil && rows.sc.getConnConfig() != nil {
rows.location = getCurrentLocation(rows.sc.cfg.Params)
}
return rows.location
Expand Down
Loading