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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
/*
** GNU Pth - The GNU Portable Threads
** Copyright (c) 1999-2002 Ralf S. Engelschall <rse@engelschall.com>
**
** This file is part of GNU Pth, a non-preemptive thread scheduling
** library which can be found at http://www.gnu.org/software/pth/.
**
** This library is free software; you can redistribute it and/or
** modify it under the terms of the GNU Lesser General Public
** License as published by the Free Software Foundation; either
** version 2.1 of the License, or (at your option) any later version.
**
** This library is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
** Lesser General Public License for more details.
**
** You should have received a copy of the GNU Lesser General Public
** License along with this library; if not, write to the Free Software
** Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
** USA, or contact Ralf S. Engelschall <rse@engelschall.com>.
**
** pth_msg.c: Pth message port facility
*/
/* ``Those who do not understand Unix
are condemned to reinvent it, poorly.''
-- Henry Spencer */
#include "pth_p.h"
#if cpp
/* message port structure */
struct pth_msgport_st {
pth_ringnode_t mp_node; /* maintainance node handle */
const char *mp_name; /* optional name of message port */
pth_t mp_tid; /* corresponding thread */
pth_ring_t mp_queue; /* queue of messages pending on port */
};
#endif /* cpp */
static pth_ring_t pth_msgport = PTH_RING_INIT;
/* create a new message port */
pth_msgport_t pth_msgport_create(const char *name)
{
pth_msgport_t mp;
/* check input */
if (name == NULL)
return_errno(NULL, EINVAL);
/* allocate message port structure */
if ((mp = (pth_msgport_t)malloc(sizeof(struct pth_msgport_st))) == NULL)
return_errno(NULL, ENOMEM);
/* initialize structure */
mp->mp_name = name;
mp->mp_tid = pth_current;
pth_ring_init(&mp->mp_queue);
/* insert into list of existing message ports */
pth_ring_append(&pth_msgport, &mp->mp_node);
return mp;
}
/* delete a message port */
void pth_msgport_destroy(pth_msgport_t mp)
{
pth_message_t *m;
/* check input */
if (mp == NULL)
return;
/* first reply to all pending messages */
while ((m = pth_msgport_get(mp)) != NULL)
pth_msgport_reply(m);
/* remove from list of existing message ports */
pth_ring_delete(&pth_msgport, &mp->mp_node);
/* deallocate message port structure */
free(mp);
return;
}
/* find a known message port through name */
pth_msgport_t pth_msgport_find(const char *name)
{
pth_msgport_t mp, mpf;
/* check input */
if (name == NULL)
return_errno(NULL, EINVAL);
/* iterate over message ports */
mp = mpf = (pth_msgport_t)pth_ring_first(&pth_msgport);
while (mp != NULL) {
if (strcmp(mp->mp_name, name) == 0)
break;
mp = (pth_msgport_t)pth_ring_next(&pth_msgport, (pth_ringnode_t *)mp);
if (mp == mpf) {
mp = NULL;
break;
}
}
return mp;
}
/* number of message on a port */
int pth_msgport_pending(pth_msgport_t mp)
{
if (mp == NULL)
return_errno(-1, EINVAL);
return pth_ring_elements(&mp->mp_queue);
}
/* put a message on a port */
int pth_msgport_put(pth_msgport_t mp, pth_message_t *m)
{
if (mp == NULL)
return_errno(FALSE, EINVAL);
pth_ring_append(&mp->mp_queue, (pth_ringnode_t *)m);
return TRUE;
}
/* get top message from a port */
pth_message_t *pth_msgport_get(pth_msgport_t mp)
{
pth_message_t *m;
if (mp == NULL)
return_errno(FALSE, EINVAL);
m = (pth_message_t *)pth_ring_pop(&mp->mp_queue);
return m;
}
/* reply message to sender */
int pth_msgport_reply(pth_message_t *m)
{
if (m == NULL)
return_errno(FALSE, EINVAL);
return pth_msgport_put(m->m_replyport, m);
}
|
