1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
|
#include <stdio.h>
#include <mongoc/mongoc-array-private.h>
#include <mongoc/mongoc-thread-private.h>
#include "mock_server/future.h"
#include "../test-libmongoc.h"
{{ header_comment }}
#define FUTURE_TIMEOUT_ABORT \
if (1) { \
fflush (stdout); \
fprintf (stderr, "%s timed out\n", BSON_FUNC); \
fflush (stderr); \
abort (); \
} else \
((void) 0)
void
future_get_void (future_t *future)
{
if (future_wait (future)) {
return;
}
FUTURE_TIMEOUT_ABORT;
}
{% for T in type_list %}
{{ T }}
future_get_{{ T }} (future_t *future)
{
if (future_wait (future)) {
return future_value_get_{{ T }} (&future->return_value);
}
FUTURE_TIMEOUT_ABORT;
}
{% endfor %}
future_t *
future_new (future_value_type_t return_type, int argc)
{
future_t *future;
future = (future_t *)bson_malloc0 (sizeof *future);
future->return_value.type = return_type;
future->argc = argc;
future->argv = (future_value_t *)bson_malloc0 ((size_t) argc * sizeof(future_value_t));
mongoc_cond_init (&future->cond);
bson_mutex_init (&future->mutex);
return future;
}
future_value_t *
future_get_param (future_t *future, int i)
{
return &future->argv[i];
}
void
future_start (future_t *future,
BSON_THREAD_FUN_TYPE (start_routine))
{
int r = mcommon_thread_create (&future->thread,
start_routine,
(void *) future);
BSON_ASSERT (!r);
}
void
future_resolve (future_t *future, future_value_t return_value)
{
bson_mutex_lock (&future->mutex);
BSON_ASSERT (!future->resolved);
BSON_ASSERT (future->return_value.type == return_value.type);
future->return_value = return_value;
future->resolved = true;
mongoc_cond_signal (&future->cond);
bson_mutex_unlock (&future->mutex);
}
bool
future_wait_max (future_t *future, int64_t timeout_ms)
{
int64_t remaining_usec = timeout_ms * 1000;
int64_t deadline = bson_get_monotonic_time () + timeout_ms * 1000;
bool resolved;
bson_mutex_lock (&future->mutex);
while (!future->resolved && remaining_usec > 0) {
mongoc_cond_timedwait (&future->cond, &future->mutex,
remaining_usec / 1000);
remaining_usec = deadline - bson_get_monotonic_time ();
}
resolved = future->resolved;
bson_mutex_unlock (&future->mutex);
if (resolved) {
future->awaited = true;
/* free memory */
mcommon_thread_join (future->thread);
}
return resolved;
}
bool
future_wait (future_t *future)
{
return future_wait_max (future, get_future_timeout_ms ());
}
void
future_destroy (future_t *future)
{
BSON_ASSERT (future->awaited);
bson_free (future->argv);
mongoc_cond_destroy (&future->cond);
bson_mutex_destroy (&future->mutex);
bson_free (future);
}
|
