Changeset: 4f7f8cf9b789 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=4f7f8cf9b789
Modified Files:
clients/Tests/exports.stable.out
clients/mapilib/mapi.c
common/stream/stream.c
gdk/gdk_atomic.h
gdk/gdk_hash.c
gdk/gdk_orderidx.c
gdk/gdk_private.h
gdk/gdk_system.c
gdk/gdk_system.h
gdk/gdk_utils.c
monetdb5/mal/mal.c
monetdb5/modules/mal/remote.c
sql/backends/monet5/sql_scenario.c
Branch: default
Log Message:
Merge with Apr2019 branch.
diffs (truncated from 1103 to 300 lines):
diff --git a/clients/Tests/exports.stable.out b/clients/Tests/exports.stable.out
--- a/clients/Tests/exports.stable.out
+++ b/clients/Tests/exports.stable.out
@@ -317,6 +317,8 @@ void *MT_thread_getdata(void);
const char *MT_thread_getname(void);
bool MT_thread_init(void);
void MT_thread_setdata(void *data);
+void MT_thread_setlockwait(MT_Lock *lock);
+void MT_thread_setsemawait(MT_Sema *sema);
void OIDXdestroy(BAT *b);
ssize_t OIDfromStr(const char *src, size_t *len, oid **dst, bool external);
ssize_t OIDtoStr(str *dst, size_t *len, const oid *src, bool external);
diff --git a/clients/mapilib/mapi.c b/clients/mapilib/mapi.c
--- a/clients/mapilib/mapi.c
+++ b/clients/mapilib/mapi.c
@@ -733,6 +733,9 @@
#define MAPIBLKSIZE 256 /* minimum buffer shipped */
+/* number of elements in an array */
+#define NELEM(arr) (sizeof(arr) / sizeof(arr[0]))
+
/* information about the columns in a result set */
struct MapiColumn {
char *tablename;
@@ -1138,7 +1141,7 @@ wsaerror(int err)
{
int i;
- for (i = 0; i < sizeof(wsaerrlist) / sizeof(wsaerrlist[0]); i++)
+ for (i = 0; i < NELEM(wsaerrlist); i++)
if (wsaerrlist[i].e == err)
return wsaerrlist[i].m;
return "Unknown error";
@@ -2237,7 +2240,7 @@ mapi_reconnect(Mapi mid)
socks[i].owner =
st.st_uid;
socks[i++].port =
atoi(e->d_name + 11);
}
- if (i == sizeof(socks) /
sizeof(socks[0]))
+ if (i == NELEM(socks))
break;
}
closedir(d);
@@ -2775,7 +2778,7 @@ mapi_reconnect(Mapi mid)
break;
case '^':
r = mid->redirects;
- m = sizeof(mid->redirects) /
sizeof(mid->redirects[0]) - 1;
+ m = NELEM(mid->redirects) - 1;
while (*r != NULL && m > 0) {
m--;
r++;
diff --git a/common/stream/stream.c b/common/stream/stream.c
--- a/common/stream/stream.c
+++ b/common/stream/stream.c
@@ -2736,7 +2736,21 @@ socket_wstream(SOCKET sock, const char *
/* streams working on an open file pointer */
#ifdef _MSC_VER
-/* special case code for reading from/writing to a Windows cmd window */
+/* special case code for reading from/writing to a Windows console and
+ * for reading from a Windows pipe
+ *
+ * For reading from and writing to the console we can use a wide
+ * character interface which means that we are independent of the code
+ * page being used. We can translate the wide characters (which are
+ * Unicode code points) easily to UTF-8.
+ *
+ * Both for reading from the console and from a pipe, we avoid hanging
+ * (waiting for input) in the read function. Instead, we only call
+ * the read function when we know there is input available. This is
+ * to prevent a deadlock situation, especially for reading from pipes,
+ * when another thread were to also interact with pipes (as happend in
+ * the scipy Python module as used in the sql/backends/monet5/pyapi05
+ * test). */
struct console {
HANDLE h;
@@ -2760,6 +2774,8 @@ console_read(stream *restrict s, void *r
if (n == 0)
return 0;
if (c->rd == c->len) {
+ while (WaitForSingleObject(c->h, INFINITE) == WAIT_TIMEOUT)
+ ;
if (!ReadConsoleW(c->h, c->wbuf, 8192, &c->len, NULL)) {
s->errnr = MNSTR_READ_ERROR;
return -1;
@@ -2914,6 +2930,58 @@ console_write(stream *restrict s, const
return (ssize_t) ((p - (const unsigned char *) buf) / elmsize);
}
+static ssize_t
+pipe_read(stream *restrict s, void *restrict buf, size_t elmsize, size_t cnt)
+{
+ HANDLE h = s->stream_data.p;
+ size_t n = elmsize * cnt;
+ unsigned char *p = buf;
+ DWORD nread;
+
+ if (h == NULL) {
+ s->errnr = MNSTR_READ_ERROR;
+ return -1;
+ }
+ if (n == 0)
+ return 0;
+ for (;;) {
+ DWORD ret = PeekNamedPipe(h, NULL, 0, NULL, &nread, NULL);
+ if (ret == 0) {
+ if (GetLastError() == ERROR_BROKEN_PIPE)
+ return 0;
+ s->errnr = MNSTR_READ_ERROR;
+ return -1;
+ }
+ if (nread > 0)
+ break;
+ Sleep(100);
+ }
+ if ((size_t) nread < n)
+ n = (size_t) nread;
+ if (!ReadFile(h, buf, (DWORD) n, &nread, NULL)) {
+ s->errnr = MNSTR_READ_ERROR;
+ return -1;
+ }
+ /* when in text mode, convert \r\n line endings to \n */
+ if (!s->binary) {
+ char *p1, *p2, *pe;
+
+ p1 = buf;
+ pe = p1 + nread;
+ while (p1 < pe && *p1 != '\r')
+ p1++;
+ p2 = p1;
+ while (p1 < pe) {
+ if (*p1 == '\r' /*&& p1[1] == '\n'*/)
+ nread--;
+ else
+ *p2++ = *p1;
+ p1++;
+ }
+ }
+ return nread / elmsize;
+}
+
static void
console_destroy(stream *s)
{
@@ -3008,26 +3076,43 @@ file_rastream(FILE *restrict fp, const c
}
}
#ifdef _MSC_VER
- if (fileno(fp) == 0 && isatty(0)) {
- struct console *c = malloc(sizeof(struct console));
- if (c == NULL) {
- destroy(s);
- return NULL;
+ if (fp == stdin) {
+ HANDLE h = GetStdHandle(STD_INPUT_HANDLE);
+
+ switch (GetFileType(h)) {
+ case FILE_TYPE_PIPE:
+ s->stream_data.p = h;
+ s->read = pipe_read;
+ s->write = NULL;
+ s->destroy = destroy;
+ s->close = NULL;
+ s->flush = NULL;
+ s->fsync = NULL;
+ s->fgetpos = NULL;
+ s->fsetpos = NULL;
+ break;
+ case FILE_TYPE_CHAR: {
+ struct console *c = malloc(sizeof(struct console));
+ if (c == NULL) {
+ destroy(s);
+ return NULL;
+ }
+ s->stream_data.p = c;
+ *c = (struct console) {
+ .h = h,
+ };
+ s->read = console_read;
+ s->write = NULL;
+ s->destroy = console_destroy;
+ s->close = NULL;
+ s->flush = NULL;
+ s->fsync = NULL;
+ s->fgetpos = NULL;
+ s->fsetpos = NULL;
+ s->isutf8 = true;
+ break;
}
- s->stream_data.p = c;
- *c = (struct console) {
- .h = GetStdHandle(STD_INPUT_HANDLE),
- };
- s->read = console_read;
- s->write = NULL;
- s->destroy = console_destroy;
- s->close = NULL;
- s->flush = NULL;
- s->fsync = NULL;
- s->fgetpos = NULL;
- s->fsetpos = NULL;
- s->isutf8 = true;
- return s;
+ }
}
#endif
return s;
diff --git a/gdk/gdk_atomic.h b/gdk/gdk_atomic.h
--- a/gdk/gdk_atomic.h
+++ b/gdk/gdk_atomic.h
@@ -64,9 +64,7 @@
#define ATOMIC_TAS(var, lck) (AO_test_and_set_full(&var) != AO_TS_CLEAR)
#define ATOMIC_ISSET(var, lck) (var != AO_TS_CLEAR)
-#else
-
-#if defined(_MSC_VER) && !defined(__INTEL_COMPILER) &&
!defined(NO_ATOMIC_INSTRUCTIONS)
+#elif defined(_MSC_VER) && !defined(__INTEL_COMPILER) &&
!defined(NO_ATOMIC_INSTRUCTIONS)
#include <intrin.h>
@@ -273,8 +271,6 @@ static inline bool
}
#define ATOMIC_ISSET(var, lck) __ATOMIC_ISSET(&var, &(lck).lock)
-#endif
-
#endif /* LIBATOMIC_OPS */
#endif /* _GDK_ATOMIC_H_ */
diff --git a/gdk/gdk_hash.c b/gdk/gdk_hash.c
--- a/gdk/gdk_hash.c
+++ b/gdk/gdk_hash.c
@@ -170,73 +170,78 @@ BATcheckhash(BAT *b)
bool ret;
lng t = 0;
- ACCELDEBUG t = GDKusec();
- MT_lock_set(&GDKhashLock(b->batCacheid));
- ACCELDEBUG t = GDKusec() - t;
+ /* we don't need the lock just to read the value b->thash */
if (b->thash == (Hash *) 1) {
- Hash *h;
- const char *nme = BBP_physical(b->batCacheid);
- int fd;
+ /* but when we want to change it, we need the lock */
+ ACCELDEBUG t = GDKusec();
+ MT_lock_set(&GDKhashLock(b->batCacheid));
+ ACCELDEBUG t = GDKusec() - t;
+ /* if still 1 now that we have the lock, we can update */
+ if (b->thash == (Hash *) 1) {
+ Hash *h;
+ int fd;
- b->thash = NULL;
- if ((h = GDKzalloc(sizeof(*h))) != NULL &&
- (h->heap.farmid = BBPselectfarm(b->batRole, b->ttype,
hashheap)) >= 0) {
- stpconcat(h->heap.filename, nme, ".thash", NULL);
+ b->thash = NULL;
+ if ((h = GDKzalloc(sizeof(*h))) != NULL &&
+ (h->heap.farmid = BBPselectfarm(b->batRole,
b->ttype, hashheap)) >= 0) {
+ const char *nme = BBP_physical(b->batCacheid);
+ stpconcat(h->heap.filename, nme, ".thash",
NULL);
- /* check whether a persisted hash can be found */
- if ((fd = GDKfdlocate(h->heap.farmid, nme, "rb+",
"thash")) >= 0) {
- size_t hdata[HASH_HEADER_SIZE];
- struct stat st;
+ /* check whether a persisted hash can be found
*/
+ if ((fd = GDKfdlocate(h->heap.farmid, nme,
"rb+", "thash")) >= 0) {
+ size_t hdata[HASH_HEADER_SIZE];
+ struct stat st;
- if (read(fd, hdata, sizeof(hdata)) ==
sizeof(hdata) &&
- hdata[0] == (
+ if (read(fd, hdata, sizeof(hdata)) ==
sizeof(hdata) &&
+ hdata[0] == (
#ifdef PERSISTENTHASH
- ((size_t) 1 << 24) |
+ ((size_t) 1 << 24) |
#endif
- HASH_VERSION) &&
- hdata[4] == (size_t) BATcount(b) &&
- fstat(fd, &st) == 0 &&
- st.st_size >= (off_t) (h->heap.size =
h->heap.free = (hdata[1] + hdata[2]) * hdata[3] + HASH_HEADER_SIZE *
SIZEOF_SIZE_T) &&
- HEAPload(&h->heap, nme, "thash", false) ==
GDK_SUCCEED) {
- h->lim = (BUN) hdata[1];
- h->type = ATOMtype(b->ttype);
- h->mask = (BUN) (hdata[2] - 1);
- h->width = (int) hdata[3];
- switch (h->width) {
- case BUN2:
- h->nil = (BUN) BUN2_NONE;
- break;
- case BUN4:
- h->nil = (BUN) BUN4_NONE;
- break;
+ HASH_VERSION) &&
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
