src/c/event-stream.h

1/* SPDX-License-Identifier: BSD-3-Clause

2 * Copied state survives transport overflow. Only the native owner records

3 * facts; the consumer receives a coherent replay before subsequent deltas. */

4struct event_fact {

5 struct event_fact *next;

6 struct wm_message message;

7 int kind;

8 int64_t id, index;

9 char *text;

10 bool queued;

11};

12struct event_replay {

13 struct wm_message message;

14 char *text;

15};

16static pthread_mutex_t event_mutex=PTHREAD_MUTEX_INITIALIZER;

17static struct event_fact *event_facts;

18static struct event_replay *event_replay;

19static size_t event_replay_count,event_replay_index;

20static bool event_resync;

21static struct wm_message event_pending[QUEUE_SIZE];

22static size_t event_pending_count;

23static char event_text[4097];

24static uint64_t event_resync_count;

26static bool resync_transient(int kind) {

27 /* Requests crossing an overflow are invalidated, rather than replayed

28 * after their native deadline. The manager can submit fresh operations. */

29 switch(kind) {

30 case 3: case 13: case 17: case 30: case 31: case 32: case 42:

31 case 65: case 70: case 75: case 77: case 80: case 84: case 85:

32 case 90: case 91: case 100: case 101: case 102: case 108:

33 case 110: case 115: case 117: case 126: case 127: return false;

34 default: return true;

35 }

36}

38static int fact_kind(int kind) {

39 switch(kind) {

40 case 2: case 5: case 6: case 8: case 9: case 10: case 14: case 15:

41 case 16: case 18: case 19: case 25: case 26: case 27: case 28:

42 case 37: case 40: case 41: case 43: case 45: case 46: case 61:

43 case 63: case 64: case 76: case 93: case 105: case 106:

44 case 107: case 116: case 123: case 128: case 130: case 131: return kind;

45 case 11: case 12: return 11;

46 default: return 0;

47 }

48}

49static bool fact_global(int kind) {

50 return kind==9 || kind==25 || kind==37 || kind==45 || kind==61;

51}

52static bool fact_collection(int kind,int removed) {

53 switch(removed) {

54 case 3: return kind==2 || kind==5 || kind==6 || kind==8 || kind==46 || kind==93 || kind==130 || kind==131;

55 case 13: return kind==10 || kind==11 || kind==14 || kind==15 || kind==16 ||

56 kind==18 || kind==19 || kind==26 || kind==27 || kind==123 || kind==128;

57 case 42: return kind==28 || kind==40 || kind==41 || kind==43;

58 case 65: return kind==63 || kind==64;

59 case 77: return kind==76;

60 case 108: return kind==105 || kind==106 || kind==107;

61 case 117: return kind==116;

62 default: return false;

63 }

64}

65static struct event_fact *find_event_fact(int kind,int64_t id,int64_t index) {

66 for(struct event_fact *fact=event_facts;fact;fact=fact->next)

67 if(fact->kind==kind && fact->id==id && fact->index==index) return fact;

68 return NULL;

69}

70static struct event_fact *record_event_fact(struct wm_message message) {

71 int kind=fact_kind(message.word[0]);

72 if(!kind) {

73 if(message.word[0]!=3 && message.word[0]!=13 && message.word[0]!=42 &&

74 message.word[0]!=65 && message.word[0]!=77 && message.word[0]!=108 && message.word[0]!=117)

75 return NULL;

76 struct event_fact **link=&event_facts;

77 while(*link) {

78 struct event_fact *fact=*link;

79 if(fact->id==message.word[1] && fact_collection(fact->kind,message.word[0])) {

80 *link=fact->next;

81 free(fact->text);

82 free(fact);

83 } else link=&fact->next;

84 }

85 return NULL;

86 }

87 int64_t id=fact_global(kind) ? 0 : message.word[1];

88 int64_t index=kind==5 ? message.word[2] : 0;

89 struct event_fact *fact=find_event_fact(kind,id,index);

90 if(!fact) {

91 fact=calloc(1,sizeof(*fact));

92 if(!fact) return NULL;

93 fact->kind=kind; fact->id=id; fact->index=index;

94 fact->next=event_facts; event_facts=fact;

95 }

96 fact->message=message;

97 return fact;

98}

99static void queue_event_locked(struct wm_message message) {

100 if(event_resync || !push(&bridge.events,message)) {

101 if(!event_resync) atomic_store(&bridge.input_resync,true);

102 atomic_fetch_add(&bridge.overflow,1);

103 event_resync=true;

104 /* Native input expires independently. Other transient replies are

105 * retained up to the bridge bound; resync invalidates excess tokens. */

106 if(!fact_kind(message.word[0]) && resync_transient(message.word[0]) && event_pending_count<QUEUE_SIZE)

107 event_pending[event_pending_count++]=message;

108 }

109}

110static void emit(struct wm_message message) {

111 pthread_mutex_lock(&event_mutex);

112 record_event_fact(message);

113 queue_event_locked(message);

114 pthread_mutex_unlock(&event_mutex);

115 wake(bridge.event_fd);

116}

117static void object_text(uint64_t id,int kind,const char *text) {

118 size_t length=text ? strnlen(text,4096) : 0;

119 /* A transport limit must not split the final UTF-8 character. */

120 while(length && ((unsigned char)text[length]&0xc0)==0x80) length--;

121 struct wm_message message={.word={kind,id,length,-2}};

122 pthread_mutex_lock(&event_mutex);

123 struct event_fact *fact=record_event_fact(message);

124 if(fact) {

125 if(!fact->text) fact->text=malloc(4097);

126 if(fact->text) {

127 if(length) memcpy(fact->text,text,length);

128 fact->text[length]=0;

129 if(!fact->queued) {

130 fact->queued=true;

131 queue_event_locked(message);

132 }

133 }

134 }

135 pthread_mutex_unlock(&event_mutex);

136 wake(bridge.event_fd);

137}

138static int fact_priority(const struct event_replay *entry) {

139 switch(entry->message.word[0]) {

140 case 5: return 0;

141 case 2: return 1;

142 case 8: case 6: case 46: case 93: return 2;

143 case 40: return 3;

144 case 28: case 41: case 43: return 4;

145 case 10: return 5;

146 case 11: case 12: return 8;

147 case 14: return 6;

148 case 63: case 105: return 9;

149 case 64: case 106: case 107: return 10;

150 case 25: case 37: case 45: case 61: return 11;

151 default: return 7;

152 }

153}

154static int compare_event_replay(const void *left,const void *right) {

155 int a=fact_priority(left),b=fact_priority(right);

156 return (a>b)-(a<b);

157}

158static bool begin_event_resync(void) {

159 size_t count=0;

160 for(struct event_fact *fact=event_facts;fact;fact=fact->next) count++;

161 /* Include the old queued transient replies before replies accumulated

162 * after overflow, preserving their order within the bounded journal. */

163 size_t capacity=count+QUEUE_SIZE+event_pending_count+2;

164 struct event_replay *replay=calloc(capacity,sizeof(*replay));

165 if(!replay) return false;

166 size_t index=1;

167 replay[0].message=(struct wm_message){.word={124,++event_resync_count,atomic_load(&bridge.overflow)}};

168 for(struct event_fact *fact=event_facts;fact;fact=fact->next) {

169 replay[index].message=fact->message;

170 if(fact->text) {

171 replay[index].text=strdup(fact->text);

172 if(!replay[index].text) {

173 for(size_t previous=0;previous<index;previous++) free(replay[previous].text);

174 free(replay);

175 return false;

176 }

177 replay[index].message.word[2]=strlen(fact->text);

178 }

179 fact->queued=false;

180 index++;

181 }

182 qsort(replay+1,count,sizeof(*replay),compare_event_replay);

183 replay[index++].message=(struct wm_message){.word={125,event_resync_count}};

184 struct wm_message message;

185 while(pop(&bridge.events,&message)) {

186 if(!fact_kind(message.word[0]) && resync_transient(message.word[0]))

187 replay[index++].message=message;

188 }

189 for(size_t pending=0;pending<event_pending_count;pending++) {

190 message=event_pending[pending];

191 if(resync_transient(message.word[0]))

192 replay[index++].message=message;

193 }

194 event_pending_count=0; event_resync=false;

195 event_replay=replay; event_replay_count=index; event_replay_index=0;

196 return true;

197}

198static bool next_copied_event(struct wm_message *message) {

199 pthread_mutex_lock(&event_mutex);

200 event_text[0]=0;

201 if(!event_replay && event_resync) begin_event_resync();

202 bool found=false;

203 if(event_replay) {

204 struct event_replay *entry=&event_replay[event_replay_index++];

205 *message=entry->message;

206 if(entry->text) {

207 snprintf(event_text,sizeof(event_text),"%s",entry->text);

208 free(entry->text);

209 }

210 if(event_replay_index==event_replay_count) {

211 free(event_replay); event_replay=NULL;

212 }

213 found=true;

214 } else if(pop(&bridge.events,message)) {

215 found=true;

216 if(message->word[3]==-2 && (message->word[0]==8 || message->word[0]==9 ||

217 message->word[0]==15 || message->word[0]==16 || message->word[0]==26 || message->word[0]==28)) {

218 struct event_fact *fact=find_event_fact(message->word[0],

219 fact_global(message->word[0]) ? 0 : message->word[1],0);

220 if(fact && fact->text) {

221 snprintf(event_text,sizeof(event_text),"%s",fact->text);

222 fact->queued=false;

223 message->word[2]=strlen(event_text);

224 }

225 }

226 }

227 pthread_mutex_unlock(&event_mutex);

228 return found;

229}

230int wm_event_text(void *bytes,int capacity) {

231 size_t length=strlen(event_text);

232 if(!bytes || capacity<0 || (size_t)capacity<length) return -1;

233 memcpy(bytes,event_text,length);

234 return length;

235}

236static void free_event_stream(void) {

237 while(event_facts) {

238 struct event_fact *fact=event_facts;

239 event_facts=fact->next;

240 free(fact->text); free(fact);

241 }

242 if(event_replay) {

243 for(size_t index=event_replay_index;index<event_replay_count;index++) free(event_replay[index].text);

244 free(event_replay); event_replay=NULL;

245 }

246}