Wednesday, October 16, 2019

Kubernetes Using Named Pipes and TCP Sockets

This post covers two things. The minor one being combining the Named Pipes code of the previous post (Kubernetes Using Named Pipes) with the Sockets code of the previous posts (Kubernetes Follow-On (Part 4) and Pseudo Visual Compiler Using Sockets).

Since this wasn't much of an accomplishment I decided to add the transmit of Broadcast messages to selected components to be designated in the Delivery.dat file. This proved a "little" more difficult. As will be commented upon.

Besides the indication which components were allowed to receive broadcast messages, the Delivery.dat file also, as time went by, had the application that supports the component added and whether the component could publish/transmit broadcast messages. This latter mainly to eventually allow any component to transmit a broadcast message and also eventually a message of a particular topic. This latter will need the receiving components to indicate which topics they wish to receive – that is, to subscribe to those topics in a separate section of the file.

Note: In the framework delivery method of older posts the methodology to support message topics and to subscribe to consume particular topics was the responsibility of the component and there was program support to capture these requests into a library. In the current approach, there is the separate Delivery.dat file where the "subscription" would take place. The implementation of topic delivery is for the future while the ability to send and receive broadcast messages (of any topic) is preparation for the delivery of particular topics to the components indicated by the Delivery.dat file.

The combination of the use of the Named Pipe and TCP Socket delivery methods was relatively minor and won't be gone into here. It did involve specifying in the Delivery.dat file which method the component expected. Both for transmit and receive. Hence, the paired component had to support the same method. Eventually, the broadcast message producer had to support both methods so delivery could be made to a component based upon which method the receiving component expected.

For broadcast messages I assumed that I would be able to use the "To" pipe name of each paired component as the pipe name for transmitting the message just as, for components using sockets, that the component's IP address and Windows sockets port could be used. This proved not to be the case – or at least I didn't find a way to get it to work.

Wild Goose Chase

For instance, I puzzled over a problem for quite some time. I had Component1 receiving the broadcast message (as it should have) but it appeared to be received via TCP Sockets instead of Named Pipes as I thought that it should have been. As can be seen in the Delivery.dat file data below, Component1 was marked in the Delivery.dat file to receive broadcast messages twice – once when it was paired with NewComponent to receive messages via Named Pipes where both components were application App1 components and a second time when paired with Component2 of App2 to receive messages via TCP Sockets.

While chasing this perceived problem I changed the broadcast message so that it indicated which communication method was going to be used. And as I thought, Component1 was receiving the message that was being sent via sockets. Either before or after, I had changed the table of components that were marked to receive broadcast messages to not include a component again if it was already in the table. So this resulted in Component1 only being in the table for its first occurrence in the Delivery.dat file where it was paired with NewComponent and to use Named Pipes. But still the problem persisted. So I chased it and chased a solution.

During this time Component7 kept receiving the broadcast message even though it wasn’t supposed to and components 2 and 3 didn’t receive the message although marked to receive it via Sockets. This I thought was strange but I concentrated on the Component1 problem.

Not until the morning of the last day did the light dawn even though it should have been staring me in the face all along. In the Delivery.dat pairings Component7 is paired with Component3 and Component1 is paired with Component2 where Component2 and Component3 were supposed to receive the broadcast message and Component1 (the source of my mystery) and Component7 were receiving the message.

For named pipes I had created the pipe name as 8toN where N was the component id of the component to receive the message. This because for the paired components the Delivery.dat file contained the basic name to be used via a pair of names – one name for sending to the opposite component and one for receiving from it where the names were reversed in the record of the other component. Since the BdComponent, a sample of a component to send broadcast messages, wasn’t to receive messages and hence not paired with each of the components to receive its messages there wasn’t the Delivery file pipe name to be used. Hence, the creation of the name; 8to1, 8to5, 8to6 for the three components marked to allow the receive of broadcast messages via named pipes where 8 was the component id of BdComponent.

However, for TCP Sockets the IP Address and Port of the Delivery file had to be used and was selected from the entries for the components to be allowed to receive broadcast messages via sockets. These were for Component2 and Component3. Due to my chasing what I perceived as the Component1 problem I had been blind to the obvious – that Component1 and Component7 were the components paired with 2 and 3. The problem being that the wrong port of the Component2 and Component3 was being selected. That is, the port that was to be used to send messages rather than receive them. Therefore, the broadcast messages were being received by the opposite component of the pair! Talk about being blind to the obvious and hence the wild goose chase.

Results

Delivery.dat file

1|Component1|Pipe|Bd|COSTCO-HP|1to4|4to1|

4|NewComponent|Pipe| |COSTCO-HP|4to1|1to4|

5|Component5|Pipe|Bd|COSTCO-HP|5to6|6to5|

6|Component6|Pipe|Bd|COSTCO-HP|6to5|5to6|

3|Component3|Socket|Bd|x1.x2.x.y|8004|8003|

7|Component7|Socket| |x1.x2.x.y|8003|8004|

2|Component2|Socket|Bd|x1.x2.x.y|8002|8001|

1|Component1|Socket|Bd|x1.x2.x.y|8001|8002|

8|BdComponent|Both| |COSTCO-HP|8to1|1to8|x1.x2.x.y|8008|8008|

where x1, x2, x and y have been substituted for the actual IP address for this blog post.

Text output at the start of such output for App1:

. . .

portions of the DeliveryTable as parsed from Delivery.dat

1 Component1 Pipe COSTCO-HP 1to4 4to1 0 0

4 NewComponent Pipe COSTCO-HP 4to1 1to4 0 0

5 Component5 Pipe COSTCO-HP 5to6 6to5 0 0

6 Component6 Pipe COSTCO-HP 6to5 5to6 0 0

3 Component3 Sock x1.x2.x.y 8004 8003

7 Component7 Sock x1.x2.x.y 8003 8004

2 Component2 Sock x1.x2.x.y 8002 8001

1 Component1 Sock x1.x2.x.y 8001 8002

8 BdComponent Sock COSTCO-HP 8to1 1to8 x1.x2.x.y 8008 8008

. . .

and the table built by the Broadcast-Client Request procedure invoked by the BdBroadcast package

Components to Receive Broadcast Messages

Com 1 Component1 COSTCO-HP 8to1

Com 5 Component5 COSTCO-HP 8to5

Com 6 Component6 COSTCO-HP 8to6

Com 3 Component3 17439

Com 2 Component2 16927

where 17439 and 16927 are the reversed byte port integers used by Sockets for ports 8004 and 8002, the first port of the pair specified for Component3 and Component2.

Problem With Using Named Pipes

There is a problem for the use of Named Pipes for broadcast messages.

With the use of broadcast messages there has to be a named pipe for each component that can receive a broadcast message and one for every component that can send a broadcast message. That is, there has to be a match between the pipe name used to transmit the message and that used to receive it.

And for every such named pipe there has to be a receive thread to wait for the message. This will quickly escalate the number of threads to be supported by the application if the same pipe name and receive thread can’t be used as for the paired messages.

I tried to send a broadcast message using the receive pipe name of a paired component such as 4to1 or 2to1. But this didn’t work. So I went to names like Bto1. But this only worked for one broadcasting component. (Note: By this time I also had a Component9 of App2 broadcasting messages.) So for a second component sending broadcast messages I had to create yet another pipe name. That is, it created a situation like that of the paired components of broadcast-component-to-receiving-component for each possible receiving component. So instead of one receive thread for the receiving component of the paired component a whole set of such pairs was required. And required that it the Delivery.dat file specify which components were going to transmit broadcast messages so that the pipe names could be generated.

Results having only one server to receive broadcast messages for a named pipe

Starting App1 before App2

Now got

BdComponent to send 29 BdComponent Message for all 2

Broadcast Client SenderData ToId 1 Bto1

NamePipe-Server Receive of 29 Index 4 Bto1 4317470

Component1 received a message: BdComponent Message for P 1 2

Transmit WriteFile OK 29

Broadcast Client SenderData ToId 5 Bto5

Transmit WriteFile OK 29

NamePipe-Server Receive of 29 Index 6 Bto5 4320654

Broadcast Client SenderData ToId 6 Bto6

Component5 received a message: BdComponent Message for P 5 2

with App1, when started first, getting the broadcast message by both components marked to receive such messages.

Also

Component1 received a message: NewComponent Message for Component1 2

NewComponent received a message: Component1 Message for NewComponent 4

Component1 received a message: Component2 Message for Component1 2

NamePipe-Server Receive of 35 Index 5 6to5 4320654

Component5 received a message: Component6 Message for Component5 2

So all the paired messages were also received.

Then I got for App2,

Component2 received a message: Component1 Message for Component2 3

Component6 received a message: BdComponent Message for P 6 2

NamePipe-Server Receive of 29 Index 4 Bto3 4318206

Component3 received a message: BdComponent Message for P 3 2

NamePipe-Server Receive of 29 Index 2 Bto2 4315754

Component2 received a message: BdComponent Message for P 2 2

NamePipe-Server Receive of 35 Index 5 5to6 4320658

Component6 received a message: Component5 Message for Component6 2

Component2 received a message: Component1 Message for Component2 4

Component7 received a message: Component3 Message for Component7 2

Component3 received a message: Component7 Message for Component3 2

So each component receives its paired messages and the broadcast messages from BdComponent but not from Component9 which sends its messages.

Then, starting App2 first App1 has the results

NamePipe-Server Receive of 35 Index 1 2to1 4317470

Component1 received a message: Component2 Message for Component1 2

NamePipe-Server Receive of 35 Index 5 6to5 4320654

Component5 received a message: Component6 Message for Component5 2

NamePipe-Server Receive of 28 Index 2 Bto1 4317470

Component1 received a message: Component9 Message for P 1 2

NamePipe-Server Receive of 28 Index 6 Bto5 4320654

Component5 received a message: Component9 Message for P 5 2

NamePipe-Server Receive of 28 Index 2 Bto1 4317470

Component1 received a message: Component9 Message for P 1 3

NamePipe-Server Receive of 28 Index 6 Bto5 4320654

Component5 received a message: Component9 Message for P 5 3

OpenReceivePipe \\.\pipe\4to1

Broadcast Callback Bto5

NamePipe-Server Receive of 35 Index 1 2to1 4317470

Component1 received a message: Component2 Message for Component1 3

NamePipe-Server Receive of 35 Index 5 6to5 4320654

Component5 received a message: Component6 Message for Component5 3

NamePipe-Server Receive of 37 Index 7 1to4 4322978

NewComponent received a message: Component1 Message for NewComponent 2

And App2 has the text output

NamePipe-Server Receive of 35 Index 7 3to7 4323110

Component7 received a message: Component3 Message for Component7 2

NamePipe-Server Receive of 35 Index 3 7to3 4318206

Component3 received a message: Component7 Message for Component3 2

NamePipe-Server Receive of 28 Index 6 Bto6 4320658

Component6 received a message: Component9 Message for P 6 2

Transmit WriteFile OK 28

Broadcast Client SenderData ToId 3 Bto3

Transmit WriteFile OK 28

NamePipe-Server Receive of 28 Index 4 Bto3 4318206

Broadcast Client SenderData ToId 2 Bto2

Component3 received a message: Component9 Message for P 3 2

NamePipe-Server Receive of 28 Index 2 Bto2 4315754

Component2 received a message: Component9 Message for P 2 2

NamePipe-Server Receive of 35 Index 1 1to2 4315754

Component2 received a message: Component1 Message for Component2 2

NamePipe-Server Receive of 35 Index 7 3to7 4323110

Component7 received a message: Component3 Message for Component7 3

NamePipe-Server Receive of 35 Index 5 5to6 4320658

Component6 received a message: Component5 Message for Component6 2

So all the paired component messages are received (although the Component5 message of App1 took quite some time) and all the broadcast messages of Component9 were received by the components of both apps.

When App1 is started first its broadcaster gets the pipes and when App2 is started first its broadcaster gets the pipes. Therefore separate servers are required for each different transmitter.

Therefore, by these results, a pipe is necessary for each broadcaster with a receiver for each pipe. This would be manageable in the configuration that I currently have with just two components sending broadcast messages. That is, just add another set of pipes for the second broadcaster. However, in the long run this is a non-starter.

That is, if any component can send a broadcast message, the number of pipes and the number of receiver threads goes up substantially. This wasn’t such a problem when messages were between applications and then forwarded to the components of interest as in the delivery frameworks of posts of the previous delivery method . For instance, three applications than required three pairs of pipes (to make them unidirectional). But for multiple components each potentially transmitting to a number of other components the use of named pipes becomes cumbersome and resource intensive.

Therefore, it seems that the use of sockets is a better choice. Instead of expanding the number of named pipes from the one extra per receiving component of the current code I returned to the use of sockets for most of broadcast message delivery leaving just the application one (App1) components to receive the messages via named pipes. This still necessitated a second broadcast pipe for each of the two App1 receiving components which will be illustrated in the code. This doesn’t place a great burden on the application resources since there are only two components that send broadcast messages.

However, for the general case where any component could send a broadcast message it would be necessary for the Delivery file to specify which components would do so enabling the code to create the necessary pipe names and receive threads.

The fix to allow the receive of Broadcast messages from multiple transmitters

As mentioned before, I had thought that I could use the same named pipe path name for a broadcast message as that for a paired component when the To component was the same component as that to receive the broadcast message. I failed to achieve that so came up with the solution of the above section where a separate pipe name was used for components that were allowed to receive broadcast messages. But it only delivered the messages of the broadcast component of the application that started first and, hence, secured the connection.

Therefore, I created in the Delivery package a table with a double dimensioned array of pipe names. The first dimension (the table rows) for the each component that could publish a broadcast message and the second dimension (the table columns) for the receiving components that used Named Pipes for the delivery method.

This table was then filled in with pipe names – a different one for each transmitter, receiver combination. Then the particular pipe name was used that matched both the transmitter and the receiver.

In the current configuration in Delivery.dat this resulted in

B08to01 B08to05 B08to06

B09to01 B09to05 B09to06

where the first row corresponds to the BdComponent of App1 and the second row to the Component9 of App2. The columns are for the 3 components (Component1, 5 and 6) that receive their messages via Named Pipes.

This could then be extended for additional components to allow them to send these unpaired component messages including those that receive paired messages. In addition, changes should be easy enough to make to allow components to register (via another .dat file or a different section of the general .dat file) for particular topics and the component that published the topic to have the Client package select the set of receiver components to which to transmit the topic message. (Shall I say I leave this as an exercise for the reader?)

However, like the Kubernetes project of GitHub, the use of TCP Sockets is the better choice since (with Windows) this only requires one port for each component rather then one pipe name for each From-To component combination.

Results having only a server to receive broadcast messages for each named pipe

With two transmitters of broadcast messages – BdComponent installed in App1 and Component9 installed in App2, App1 got

Component1 received a message: BdComponent Message for P 1 2 ß

Transmit sending 35 bytes WriteFile

Transmit WriteFile OK 37 4to1

NamePipe-Server Receive of 37 Index 4 4to1 4321894

Component1 received a message: NewComponent Message for Component1 2

Transmit WriteFile OK 29

Broadcast Client SenderData ToId 5

Transmit WriteFile OK 35 5to6

Transmit WriteFile OK 29

Broadcast Client SenderData ToId 6

NamePipe-Server Receive of 29 Index 6 4325078

Transmit WriteFile OK 29

Component5 received a message: BdComponent Message for P 5 2 ß

Broadcast Client SenderData ToId 3

Transmit WriteFile OK 29

Broadcast Client SenderData ToId 2

Transmit WriteFile OK 29

Broadcast Callback B08to05

Broadcast Callback B08to01

Component1 to send to Component2 35 Component1 Message for Component2 4

Client Transmit Index 1

Transmit sending 35 bytes WriteFile

Transmit WriteFile OK 35 1to2

NamePipe-Server Receive of 28 Index 3 4321894

Component1 received a message: Component9 Message for P 1 5 ß

NamePipe-Server Receive of 35 Index 5 6to5 4325078

Component5 received a message: Component6 Message for Component5 5

NamePipe-Server Receive of 35 Index 1 2to1 4321894

Component1 received a message: Component2 Message for Component1 5

NamePipe-Server Receive of 28 Index 7 4325078

Component5 received a message: Component9 Message for P 5 5 ß

So the broadcast messages now received from both transmitters of broadcast messages.

And App2 received

Component6 received a message: BdComponent Message for P 6 2 ß

NamePipe-Server Receive of 29 Index 5 4322630

NamePipe-Server Receive of 29 Index 2 4320178

Component3 received a message: BdComponent Message for P 3 2 ß

Component2 received a message: BdComponent Message for P 2 2 ß

. . .

Component6 received a message: Component9 Message for P 6 5 ß

Component7 received a message: Component3 Message for Component7 5

NamePipe-Server Receive of 28 Index 6 4322630

Component3 received a message: Component9 Message for P 3 5 ß

Broadcast Client SenderData ToId 2

Transmit WriteFile OK 28

NamePipe-Server Receive of 28 Index 3 4320178

Component2 received a message: Component9 Message for P 2 5 ß

So each component that was allowed to receive broadcast messages received them from each of the transmitters.

Delivery.dat that was used to produce the above results

In order to concentrate on the changes needed to deliver broadcast messages via Named Pipes the Delivery data file was changed to only specify the use of pipes.

The fields of each record are separated by the | token. The fields in order are

1 Component Id and Component Name

2 Application Id

3 The delivery method

4 Whether the component can receive broadcast messages

5 Whether the component will transmit broadcast messages

6 Computer name to be used for Named Pipe delivery

7 Pipe name to transmit

8 Pipe name to receive

For components that transmit broadcast messages they must support both named pipes and sockets so they can do so via sockets if the receiving component specifies the use of sockets. The IP address isn't identified.

1|Component1|1|Pipe|Bd| |COSTCO-HP|1to4|4to1|

4|NewComponent|1|Pipe| | |COSTCO-HP|4to1|1to4|

5|Component5|1|Pipe|Bd| |COSTCO-HP|5to6|6to5|

6|Component6|2|Pipe|Bd| |COSTCO-HP|6to5|5to6|

3|Component3|2|Pipe|Bd| |COSTCO-HP|3to7|7to3|

7|Component7|2|Pipe| | |COSTCO-HP|7to3|3to7|

2|Component2|2|Pipe|Bd| |COSTCO-HP|2to1|1to2|

1|Component1|1|Pipe|Bd| |COSTCO-HP|1to2|2to1|

8|BdComponent|1|Both| |Bd|COSTCO-HP|8to1|1to8|xx.x.x.xxx|8008|8008|

9|Component9|2|Both| |Bd|COSTCO-HP|9to1|1to9|xx.x.x.xxx|8009|8009|

Notice that Component1 and Component5 of application 1 and Component6, Component3, and Component2 of application 2 are specified to receive broadcast messages while BdComponent of application 1 and Component9 of application 2 transmit broadcast messages.

Windows 10 Problems

What with the support for Windows 7 ending at the end of the year I went ahead and got a Windows 10 laptop. What a fiasco that has turned out to be.

It won't allow the applications that I have been using to be installed or to run. And the GNAT GPS Ada that I have been using for these private projects won't run in the versions that I have been using and the latest version GNAT GPS 2019 that I had to install as the only one seemingly available for download has problems that prevent it from being usable. For instance, I did manage to get a build for the App1 application but it failed for the App2 application that is identical in structure. That is, it built to a much smaller executable that doesn't do anything. Perhaps I'll figure that one out but I failed to do so after a couple of days of trying.

Then there is the need for special treatment to be able to run my own created applications (that is, app1 and app2). That I found out about and can get around. But the need to buy all new applications suitable for Windows 10 seems to me to be a RIP OFF. A version of Windows that Microsoft is pushing us into needing to use that is completely deficient. That is, Windows 10 can't install or run programs that earlier versions of Windows had no problems running.

So I'll need to continue to use Windows 7 even though it’s unsupported come the end of the year.

Return to the use of Sockets for delivery

Returning to the use of Win Sockets for delivery to three of the App2 components while retaining the use of Named Pipes for the App1 components and one of the App2 components the Delivery.dat file looks like

1|Component1|1|Pipe|Bd| |COSTCO-HP|1to4|4to1|

4|NewComponent|1|Pipe| | |COSTCO-HP|4to1|1to4|

5|Component5|1|Pipe|Bd| |COSTCO-HP|5to6|6to5|

6|Component6|2|Pipe|Bd| |COSTCO-HP|6to5|5to6|

3|Component3|2|Socket|Bd| |xx.x.x.xxx|8004|8003|

7|Component7|2|Socket| | |xx.x.x.xxx|8003|8004|

2|Component2|2|Socket|Bd| |xx.x.x.xxx|8002|8001|

1|Component1|1|Socket|Bd| |xx.x.x.xxx|8001|8002|

8|BdComponent|1|Both| |Bd|COSTCO-HP|8to1|1to8|xx.x.x.xxx|8008|8008|

9|Component9|2|Both| |Bd|COSTCO-HP|9to1|1to9|xx.x.x.xxx|8009|8009|

With the use of Sockets the same port could be used to transmit to a component and the same receive callback thread could be used whether the receiving component was receiving a paired component message or a broadcast message.

This is important since I had to have a different named pipe for each From component to To component connection. So I had to have a different receive thread to await the message for each connection. Therefore, for larger applications with more components this would become a major use of resources.

I had thought that I could find a method to reuse the paired component connection for the broadcast messages. But as I have mentioned I failed to do so. I suppose I could try again but to what purpose when the use of Sockets can be used in place of Named Pipes.

Results using both named pipes and sockets to receive messages

The App1 results

Component1 received a message: NewComponent Message for Component1 1 ß the paired

. . .

NewComponent received a message: Component1 Message for NewComponent 2 ßcomponents

. . .

Component5 received a message: BdComponent Message for P 5 2 ß the local broadcast

NamePipe-Server Receive of 29 Index 2 4324078

Component1 received a message: BdComponent Message for P 1 2 ß messages

. . .

Component1 received a message: Component2 Message for Component1 1 ß remote paired

. . .

Component1 received a message: Component9 Message for P 1 2 ß the remote broadcast

NamePipe-Server Receive of 28 Index 6 4327262

Component5 received a message: Component9 Message for P 5 2 ß messages

NamePipe-Server Receive of 35 Index 4 6to5 4327262

Component5 received a message: Component6 Message for Component5 2 ß remote paired

So the App1 components received all their messages. Note these all used Named Pipes.

The App2 results

Component6 received a message: Component5 Message for Component6 2 ç remote paired

Socket-Server Receive of 29 Index 2

Component3 received a message: BdComponent Message for S 3 1 ç remote broadcast

Socket Server Receive after Listen 2 True

Socket-Server Receive of 29 Index 1

Component2 received a message: BdComponent Message for S 2 1 ç remote broadcast

. . .

Component2 received a message: Component1 Message for Component2 2 ß remote paired

. . .

Component6 received a message: Component5 Message for Component6 3

NamePipe-Server Receive of 29 Index 2 4327266

Component6 received a message: BdComponent Message for P 6 2 ß all three

Socket-Server Receive of 29 Index 2

Component3 received a message: BdComponent Message for S 3 2 ß components rec'd

Socket Server Receive after Listen 2 True

Socket-Server Receive of 29 Index 1

Component2 received a message: BdComponent Message for S 2 2 ß remote broadcast

Socket Server Receive after Listen 1 True

Socket-Server Receive of 35 Index 1

Component2 received a message: Component1 Message for Component2 3

. . .

Component7 received a message: Component3 Message for Component7 1 ß local paired

. . .

Component3 received a message: Component7 Message for Component3 1 ß local paired

Transmit sent using client socket port 17439

ERROR: Write to pipe failed 404 0

. . .

Component3 received a message: Component9 Message for S 3 1 ß local broadcast

Broadcast Callback B09to06

OpenReceivePipe \\.\pipe\B09to06

NamedPipe Server (Receive) Handle is Valid 412

Server connecting to client...

Socket Server Receive after Listen 2 True

Client Socket Connected to Transmit

412

NamedPipe Server setting Connected 0

Socket-Server Receive of 28 Index 1

Component2 received a message: Component9 Message for S 2 1 ß local broadcast

. . .

Component6 received a message: Component9 Message for P 6 2 ß local broadcast

So the App2 components received all their messages. Component6 via Named Pipes and Components 3, 7, and 2 via Sockets.

Application Code

The code will be resented in a separate post.

Monday, July 29, 2019

Kubernets Using Named Pipes

Under a heading of Comments in the "Kubernetes Follow-On (Part 4)" post I noted

And oh my; the reduction in code. As can be seen in earlier posts such as using Named Pipes where I was sending messages between applications and then having the application figure out to what component(s) to deliver the message.

Although, thinking about it now, it would seem that the use of named pipes could achieve the same result. Just like the TCP sockets require a port for every component, the use of pipes must be able to use a pipe name for every component rather than every application. Something to look into to see what would need to be done and whether there would be a similar reduction in the amount of code.

This post is the result of structuring the communications using named pipes similar to that of the TCP sockets of that post – as well as the "Pseudo Visual Compiler Using Sockets" post that followed it.

Once again, the code is so much smaller than my previous "framework" based applications that used named pipes to communicate between applications rather than directly between components of the applications as occurred to me when I had a brief look see at the Kubernets approach.

Overall Approach

As with TCP sockets, there is a Delivery.dat file to specify the pairs of components that can communicate.

For Named Pipes the records of the file contain each component's numeric identifier and name. The component identification is followed by the name of the PC on which it resides and the client and server pipe names associated with the component. In addition, looking forward to when this method will be extended to allow the communications to be either by named pipes or by sockets, the component identifier is followed by a field denoting the communications method to be used – Pipe or Socket. I will attempt to extend the approach to allow either communication method next.

As with TCP sockets, an application is made up a set of components and the implementation of the common communications method – in this case that to connect the pair of components and to send and receive messages between the two. Along with support packages including the Threads package to create a thread for each component and one to wait to receive its messages. As with TCP sockets, the transmit is done via the component's thread.

Other support packages are

1) Delivery to locate, parse, validate, and create an internal table of the component pairs.

2) Directory to find the path to the Delivery.dat file.

3) ExecItf to interface with the executive; that is, Windows.

4) Itf to specify commonly used Ada types.

5) CStrings to support C-type string functions.

6) TextIO to be able to output a string to the console that includes conversions (type casting) from integers, etc as one combined string to prevent the inter-mixture of the output caused by thread suspension during output.

Except for Delivery and a minor change to Threads, the support packages are the same as in previous posts.

In addition there is the Choose package that is being used to direct the interface between the components and the communication support method to the Named Pipe packages. This was added in anticipation of the next step of allowing both named pipe and socket communications. Currently it only forwards the component requests to the NamedPipe package and the Client and Server subpackages.

Implementation

As with any Ada project there is the main procedure that is invoked when the application is launched. For this post these consist of App1.adb and App2.adb for the two applications. One application (App1) has two components (Component1 and NewComponent) that intra communicate with each other and one component (Component5) that communicates with Component6 of the second application.

As before, these main procedures initiate the application. For App1

with Component1;

with Component5;

with NewComponent;

with Delivery;

with NamedPipe;

with Threads;

with Text_IO;

procedure App1 is

package Int_IO is new Text_IO.Integer_IO( Integer );

function to_Int is new Unchecked_Conversion

( Source => System.Address,

Target => Integer );

begin -- App1

Text_IO.Put("App5");

Int_IO.Put(to_Int(ExecItf.GetCurrentThread)); -- current thread

Text_IO.Put_Line(" ");

-- Locate, Parse, and Validate the Delivery.dat file

Delivery.Initialize;

NamedPipe.Initialize;

-- Install the components of this application

Component1.Install;

Component5.Install;

NewComponent.Install;

-- Now, after all the components have been installed, create their threads

Threads.Create;

end App1;

As usual, the main procedure does the initialization (where that of Delivery parses the Delivery.dat file), gets the components started, and then creates the threads requested by the components and their interfaces to the named pipe code. Threads Create never returns and the code runs in the various threads after that.

App2 is the same except for only doing Component6.Install rather than that of the three components.

Component1

(spec)

with ExecItf;

package Component1 is

-- Return component's wakeup event handle

function WakeupEvent

return ExecItf.HANDLE;

procedure Install;

end Component1;

(body)

with Choose;

with Itf;

with NamedPipe.Server;

with System;

with TextIO;

with Text_IO;

with Threads;

with Unchecked_Conversion;

package body Component1 is

ComponentWakeup

-- Wakeup Event handle of the component

: ExecItf.HANDLE;

CurrentThread

: ExecItf.HANDLE;

MessagesSent

: Integer := 0;

From4 : Boolean; -- Component1 from NewCompoent message

To4 : Boolean; -- Component1 to NewCompoent message

procedure Callback

( Id : in Integer

);

-- Receive Message by Component1

procedure ReceiveCallback

( Message : in Itf.BytesType

);

procedure Install is

Result : Threads.RegisterResult;

use type Threads.InstallResult;

function to_Callback is new Unchecked_Conversion

( Source => System.Address,

Target => Threads.CallbackType );

function to_RecvCallback is new Unchecked_Conversion

( Source => System.Address,

Target => Choose.ReceiveCallbackType );

begin -- Install

CurrentThread := ExecItf.GetCurrentThread;

-- Install the component into the Threads package.

Result := Threads.Install

( Name => "Component1",

Index => Threads.TableCount + 1,

Priority => Threads.NORMAL,

Callback => to_Callback(Callback'Address)

);

if Result.Status = Threads.VALID then

ComponentWakeup := Result.Event;

-- Request the ability to send to NewComponent.

To4 := Choose.Request( Choose.Client,

"Component1",

"NewComponent",

4 );

if not To4 then

Text_IO.Put_Line(

"Client not valid for Component1, NewComponent pair" );

end if;

-- Request the ability to receive from NewComponent.

From4 := Choose.Request

( Choose.Server,

"Component1",

"NewComponent",

to_RecvCallback(ReceiveCallback'Address) );

if not From4 then

Text_IO.Put_Line(

"Server not valid for Component1, NewComponent pair" );

end if;

end Install;

-- Return component's wakeup event handle

function WakeupEvent

return ExecItf.HANDLE is

begin -- WakeupEvent

return ComponentWakeup;

end WakeupEvent;

-- Received message from another component

procedure ReceiveCallback

( Message : in Itf.BytesType

) is

Text : Itf.V_80_String_Type;

begin -- ReceiveCallback

Text.Data(1..30) := "Component1 received a message:";

declare

-- Assume string for entire message

Msg : String(1..Message.Count);

for Msg use at Message.Bytes'Address;

begin

Text := TextIO.Concat(Text.Data(1..30),Msg);

TextIO.Put_Line(Text);

end;

end ReceiveCallback;

function toDigit

( Num : in Integer

) return Character is

-- Convert number to byte digit from 0 thru 9.

Number : Itf.Byte;

Digit : Character;

for Digit use at Number'Address;

begin -- toDigit

Number := Itf.Byte(Num Mod 10 + 48); -- 48 is '0'

return Digit;

end toDigit;

-- Forever loop as initiated by Threads

procedure Callback

( Id : in Integer

) is

Message : Itf.BytesType;

Text : Itf.V_80_String_Type;

ThreadHandle : ExecItf.HANDLE;

begin -- Callback

Text.Data(1..22) := "in Component1 callback";

Text := TextIO.Concat(Text.Data(1..22), Id);

TextIO.Put_Line(Text);

ThreadHandle := ExecItf.GetCurrentThread;

loop -- forever

if To4 then

Text.Data(1..34) := "Component1 to send to NewComponent";

declare

Msg : String := "Message for NewComponent ";

for Msg use at Message.Bytes'address;

begin

Message.Count := Msg'Length;

MessagesSent := MessagesSent + 1;

Msg(Message.Count) := ToDigit(MessagesSent);

Text := TextIO.Concat(Text.Data(1..34),Message.Count);

Text := TextIO.Concat(Text.Data(1..Text.Count),Msg(1..Message.Count));

TextIO.Put_Line(Text);

if not Choose.Transmit( 1, 4, -- from 1 (Component1) to 4 (NewComponent)

Message )

then

Text_IO.Put_Line( "Message not sent to NewComComponent" );

end if;

end;

end if;

Delay(1.0);

end loop;

end Callback;

end Component1;

Component5, 6, and NewComponent are directly similar.

Each component's Install (as invoked from its main procedure) installs itself with Threads to get its Callback thread to run in. The Callback procedure will begin running after Threads Create of the main procedure is invoked. Next it installs itself with the communication packages via the Choose intermediate interface. The first invocation is to the Client package to set up to transmit to the second component of the pair – ("NewComponent", 4 in this case). The second invocation is to the Server package to set up to receive from the second component of the pair. The last parameter passes the location of the procedure to be passed the messages that are received.

Although not shown in any of these example components, other pairs could also be installed where the first component of each pair would be the component that's making invocation. Also, of course, if the component is only going to transmit to the "To" component only the first invocation is needed. And if only to receive, only the "From" Request is needed.

The ReceiveCallback procedure is entered via the particular Server Callback thread that received the message. Therefore, it runs in that thread rather than the Callback thread of the component. If it invoked procedures or functions that were also called by the Callback thread, these procedures and functions would need to only reference stack variables to protect against the second thread gaining control in the middle of the update of a static variable before the first had finished thus causing garbage results.

The Callback procedure is invoked from Threads and executes in its own thread. It Transmits a message every second to NewComponent with a new trailing message counter each time. The other components are similar except for sending the message to their own paired component.

In a real application, the received message would be validated. And cause particular actions to be taken. Some of which might be to cause messages to be sent to particular components.

Delivery

Delivery is like in the past except that the parse has to determine which communication method the Delivery.dat records specify to be used. In preparation for also including TCP Sockets the records can specify Pipe or Socket and the DeliveryTable has been changed to be able to store the component data either way.

The record format is either

if Method is Pipe or

if the Method is Socket.

The DeliveryTable has been expanded to have records that contain

type DeliveryTableDataType

is record

ComId : Choose.ComponentIdsType;

-- Identifier of component of the table entry

ComName : Choose.ComponentNameType;

-- Name of component of the table entry

Partner : LocationType;

-- Index of the component with the opposite ports

Usage : Choose.UsageType;

-- How to interpret the rest of the fields

ComputerName : Choose.ComponentNameType;

-- Identifier of the Client PC for Usage of Pipe

PipeServer : NamedPipe.PipeNameType;

-- Short name of the server/receive pipe for Usage of Pipe

PipeClient : NamedPipe.PipeNameType;

-- Short name of the client/transmit pipe for Usage of Pipe

PCAddress : BytesType;

-- IP address of PC of the component for Usage of Socket

PortServer : Integer;

-- Identifier of the server/receive port for Usage of Socket

PortClient : Integer;

-- Identifier of the client/transmit port for Usage of Socket

end record;

where the Usage field contains the communication Method. Either ComputerName, PipeServer, and PipeClient or PCAddress, PortServer, and PortClient will be filled in.

I'll wait until the next post that allows the use of either method to provide the Delivery code. The current Delivery.dat file contents are

1|Component1|Pipe|COSTCO-HP|1to4|4to1|

4|NewComponent|Pipe|COSTCO-HP|4to1|1to4|

5|Component5|Pipe|COSTCO-HP|5to6|6to5|

6|Component6|Pipe|COSTCO-HP|6to5|5to6|

where the first three records refer to App1 components and the fourth is an App2 component.

It is anticipated that the ComputerName will be needed if the pipes are to communicate between different PCs. For this test only one PC was involved so the ComputerName wasn't used – . was used instead.

Common Named Pipe Packages

Choose

Choose is in anticipation of communicating via Named Pipes or Sockets depending on which method is specified by the Delivery.dat file. However, nothing has been added as yet to forward the component request to the socket client or server.

(spec)

with Itf;

with Threads;

package Choose is

subtype ComponentIdsType

-- Identifier of the hosted components.

-- Notes:

-- This allows for a configuration with a maximum of 63 components.

is Integer range 0..63;

type ComponentNameType

-- Name of the hosted components

is record

Count : Integer; -- number of characters in name

Value : String(1..20);

end record;

type OptionType

is ( Client,

Server );

type UsageType

is ( Pipe,

Socket );

type ReceiveCallbackType

-- Callback to return received message to its component

is access procedure( Message : in Itf.BytesType );

function Request

-- Request a Client component pairing

( Option : in OptionType := Client;

FromName : in String;

FromId : in ComponentIdsType;

ToName : in String;

ToId : in ComponentIdsType

) return Boolean;

function Request

-- Request a Server component pairing

( Option : in OptionType := Server;

FromName : in String;

FromId : in ComponentIdsType;

ToName : in String;

ToId : in ComponentIdsType;

RecvCallback : in ReceiveCallbackType

) return Boolean;

function Transmit

-- Request to Client to transmit message

( FromId : in ComponentIdsType;

ToId : in ComponentIdsType;

Message : in Itf.BytesType

) return Boolean;

end Choose;

(body)

with NamedPipe.Client;

with NamedPipe.Server;

with TextIO;

package body Choose is

function Request

( Option : in OptionType := Client;

FromName : in String;

FromId : in ComponentIdsType;

ToName : in String;

ToId : in ComponentIdsType

) return Boolean is

begin -- Request

if Option = Client then

return NamedPipe.Client.Request

( FromName => FromName,

FromId => FromId,

ToName => ToName,

ToId => ToId );

else

declare

Text : Itf.V_80_String_Type;

begin

Text.Data(1..37) := "ERROR: Needs to have message callback";

Text.Count := 37;

TextIO.Put_Line( Text );

end;

return False;

end if;

end Request;

function Request

( Option : in OptionType := Server;

FromName : in String;

FromId : in ComponentIdsType;

ToName : in String;

ToId : in ComponentIdsType;

RecvCallback : in ReceiveCallbackType

) return Boolean is

begin -- Request

if Option = Client then

return NamedPipe.Client.Request

( FromName => FromName,

FromId => FromId,

ToName => ToName,

ToId => ToId );

else

return NamedPipe.Server.Request

( FromName => FromName,

FromId => FromId,

ToName => ToName,

ToId => ToId,

RecvCallback => RecvCallback );

end if;

end Request;

function Transmit

( FromId : in ComponentIdsType;

ToId : in ComponentIdsType;

Message : in Itf.BytesType

) return Boolean is

begin -- Transmit

return NamedPipe.Client.Transmit

( FromId => FromId,

ToId => ToId,

Message => Message );

end Transmit;

end Choose;

When fully implemented, the method will need to be determined so that the Socket functions can be invoked when appropriate.

NamedPipe

(spec)

with Choose;

package NamedPipe is

-- parent of Client and Server packages

type PipeNameType

is record

Count : Integer; -- number of characters in name

Value : String(1..20);

end record;

type DataType

is record

Connected : Boolean; -- True equals connected to server

end record;

type ListType

is array (1..Choose.ComponentIdsType'Last) of DataType;

type DataListType

is record

List : ListType;

end record;

Data

: DataListType;

procedure Initialize;

end NamedPipe;

(body)

package body NamedPipe is

procedure Initialize is

begin -- Initialize

for I in 1..Choose.ComponentIdsType'Last loop

Data.List(I).Connected := False;

end loop;

end Initialize;

end NamedPipe;

NamedPipe-Client

(spec)

with Choose;

with Itf;

package NamedPipe.Client is

-- child package of NamedPipe

function Request

-- Request a Client component pairing

( FromName : in String;

FromId : in Choose.ComponentIdsType;

ToName : in String;

ToId : in Choose.ComponentIdsType

) return Boolean;

function Transmit

( FromId : in Choose.ComponentIdsType;

ToId : in Choose.ComponentIdsType;

Message : in Itf.BytesType

) return Boolean;

end NamedPipe.Client;

(body)

with Delivery;

with Interfaces.C;

with ExecItf;

with System;

with TextIO;

with Text_IO;

with Threads;

with Unchecked_Conversion;

package body NamedPipe.Client is

-- child package of NamedPipe

package Int_IO is new Text_IO.Integer_IO( Integer );

-- SocketClient Sender Data

type SenderDataType

is record

FromName : Choose.ComponentNameType; -- Name and Id of the invoking component

FromId : Choose.ComponentIdsType; -- from which message is to be sent

ToId : Choose.ComponentIdsType; -- Name and Id of remote component

ToName : Choose.ComponentNameType; -- to be sent the message

ThreadId : Integer; -- Id of Transmit thread

ComputerName : Choose.ComponentNameType;

DelPipeName : NamedPipe.PipeNameType; -- pipename

PipeName : NamedPipe.PipeNameType; -- must be of the form \\.\pipe\pipename

Created : Boolean; -- True if Transmit pipe opened

Sender : ExecItf.HANDLE; -- Pipe handle

end record;

type SenderListType

is array (1..Choose.ComponentIdsType'Last) of SenderDataType;

type SenderType

is record

Count : Choose.ComponentIdsType;

List : SenderListType;

end record;

SenderData

: SenderType;

function OpenTransmitPipe

( Index : in Choose.ComponentIdsType

) return Boolean;

-- Request a Client component pairing

function Request

( FromName : in String;

FromId : in Choose.ComponentIdsType;

ToName : in String;

ToId : in Choose.ComponentIdsType

) return Boolean is

Index : Integer;

MatchIndex : Delivery.LocationType;

Partner : Delivery.LocationType;

PipeSize : Integer;

use type Delivery.LocationType;

begin -- Request

if SenderData.Count < Threads.MaxComponents then

Index := SenderData.Count + 1;

SenderData.Count := Index;

SenderData.List(Index).FromName.Count := FromName'Length;

SenderData.List(Index).FromName.Value(1..FromName'Length) := FromName;

SenderData.List(Index).FromId := FromId;

SenderData.List(Index).ToName.Count := ToName'Length;

SenderData.List(Index).ToName.Value(1..ToName'Length) := ToName;

SenderData.List(Index).ToId := ToId;

SenderData.List(Index).Created := False;

-- Find the partner in DeliveryTable. This is a validation as

-- well as that the invocating component is correct that the from

-- and to component ids and names match the table.

MatchIndex := Delivery.Lookup(Choose.Pipe, FromId, ToId);

-- Set the Computer Name and the pipes.

if MatchIndex > 0 then

Partner := Delivery.Partner( MatchIndex );

-- Fill in computer name and pipe

SenderData.List(Index).ComputerName := Delivery.ComputerName(Partner);

SenderData.List(Index).DelPipeName := Delivery.Pipe(Partner);

PipeSize := SenderData.List(Index).DelPipeName.Count;

SenderData.List(Index).PipeName.Value(1..9) := "\\.\pipe\";

for I in 1..PipeSize loop

SenderData.List(Index).PipeName.Value(9+I) :=

SenderData.List(Index).DelPipeName.Value(I);

end loop;

SenderData.List(Index).PipeName.Count := PipeSize + 9;

SenderData.List(Index).PipeName.Value(9+PipeSize+1) := ASCII.NUL;

Text_IO.Put( "MatchIndex " );

Int_IO.Put( Integer(MatchIndex) );

Text_IO.Put( " " );

Text_IO.Put( " ClientPipe " );

Text_IO.Put( SenderData.List(Index).PipeName.Value(1..9+PipeSize) );

Text_IO.Put_Line( " " );

else

Text_IO.Put_Line( "ERROR: From-To not valid for Client" );

return False;

end if;

Text_IO.Put( "SenderData count " );

Int_IO.Put( FromId );

Int_IO.Put( SenderData.Count );

Text_IO.Put_Line( " " );

return True;

else

Text_IO.Put_Line( "ERROR: Too many Senders" );

return False;

end if;

end Request;

function Lookup

( FromId : in Choose.ComponentIdsType;

ToId : in Choose.ComponentIdsType

) return Choose.ComponentIdsType is

begin -- Lookup

for I in 1..SenderData.Count loop

if SenderData.List(I).FromId = FromId and then

SenderData.List(I).ToId = ToId

then

return I;

end if;

end loop;

return 0;

end Lookup;

-- Open the Transmit Pipe

function OpenTransmitPipe

( Index : in Choose.ComponentIdsType

) return Boolean is

Count : Integer;

use type Interfaces.C.unsigned_long;

use type System.Address;

function AddrToLPSCSTR -- convert address to ExecItf pointer

is new Unchecked_Conversion( Source => System.Address,

Target => ExecItf.LPCSTR );

function to_Int is new Unchecked_Conversion

( Source => System.Address,

Target => Integer );

begin -- OpenTransmitPipe

Count := SenderData.List(Index).PipeName.Count;

if Count > 0 then

SenderData.List(Index).Sender :=

ExecItf.CreateNamedPipe

( Name => AddrtoLPSCSTR(SenderData.List(Index).PipeName.Value'Address),

OpenMode => ExecItf.PIPE_ACCESS_DUPLEX or

ExecItf.FILE_FLAG_OVERLAPPED,

PipeMode => ExecItf.PIPE_TYPE_MESSAGE or --16#00000004#

ExecItf.PIPE_READMODE_MESSAGE or --2 (0 is BYTE)

ExecItf.PIPE_WAIT, --0

MaxInstances => 1,

OutBufferSize => Interfaces.C.unsigned_long(Itf.MessageSize), -- output buffer size

InBufferSize => Interfaces.C.unsigned_long(Itf.MessageSize), -- input buffer size

DefaultTimeOut => ExecItf.PIPE_TIMEOUT,

SecurityAttributes => null ); -- default priority attributes --

if SenderData.List(Index).Sender /= ExecItf.Invalid_Handle_Value then

SenderData.List(Index).Created := True;

-- Wait for server to connect

declare

Text : Itf.V_80_String_Type;

begin

Text.Data(1..40) := "Client Transmit connected for remote app";

Text := TextIO.Concat(Text.Data(1..40),to_Int(SenderData.List(Index).Sender));

Text :=

TextIO.Concat(Text.Data(1..Text.Count),

SenderData.List(Index).PipeName.Value(1..Count));

TextIO.Put_Line(Text);

end;

NamedPipe.Data.List(Index).Connected := True;

end if;

return SenderData.List(Index).Created;

else -- error creating pipe

Text_IO.Put_Line("OpenTransmitPipe Client Handle invalid");

-- close the handle -- first check if pipeClient is non null

return False;

end if;

end OpenTransmitPipe;

function Transmit

( FromId : in Choose.ComponentIdsType;

ToId : in Choose.ComponentIdsType;

Message : in Itf.BytesType

) return Boolean is

BytesWritten

-- Number of bytes sent

: ExecItf.INT;

Index : Choose.ComponentIdsType;

Opened : Boolean;

Text : Itf.V_80_String_Type;

begin -- Transmit

-- Note: Transmit runs in the thread of the sending component. Therefore,

-- this Transmit runs in a different thread from other Clients even

-- if the sending component sends to multiple remote components since

-- it won't send to a different component until Transmit returns.

if Message.Count = 0 then

return False;

end if;

Index := Lookup( FromId, ToId );

Text.Data(1..21) := "Client Transmit Index";

Text := TextIO.Concat(Text.Data(1..21), Integer(Index));

TextIO.Put_Line(Text);

if Index <= 0 then

return False;

end if;

-- The sender always starts up on the localhost.

if not SenderData.List(Index).Created then

Text.Data(1..23) := "Client Transmit do Open";

Opened := OpenTransmitPipe(Index);

if not Opened then

Text := TextIO.Concat(Text.Data(1..23),"NOT Opened");

TextIO.Put_Line(Text);

return False;

end if;

TextIO.Put_Line(Text);

end if;

-- Create a client handle and connect it to the remote

declare -- try

Status : ExecItf.BOOL;

use type Interfaces.C.unsigned_long;

function to_Int is new Unchecked_Conversion

( Source => System.Address,

Target => Integer );

function toLPDWORD -- convert address to Exec_Itf pointer

is new Unchecked_Conversion( Source => System.Address,

Target => ExecItf.LPDWORD );

function toLPVOID -- convert address to Exec_Itf pointer

is new Unchecked_Conversion( Source => System.Address,

Target => ExecItf.LPVOID );

begin

Text.Data(1..16) := "Transmit sending";

Text := TextIO.Concat(Text.Data(1..16),Integer(Message.Count));

Text := TextIO.Concat(Text.Data(1..Text.Count),"bytes");

if not SenderData.List(Index).Created then

Text := TextIO.Concat(Text.Data(1..Text.Count),"Not Created");

TextIO.Put_Line(Text);

return False;

else

Text := TextIO.Concat(Text.Data(1..Text.Count),"WriteFile");

TextIO.Put_Line(Text);

end if;

-- Send message via the component's thread.

Status := ExecItf.WriteFile

( File => SenderData.List(Index).Sender, -- pipe handle

Buffer => toLPVOID(Message.Bytes'address), -- buffer to write from

NumberOfBytesToWrite => ExecItf.ULONG(Message.Count),

NumberOfBytesWritten => toLPDWORD(BytesWritten'address),

Overlapped => null ); -- not overlapped I/O

if Integer(Status) > 0 then -- Write successful

if Integer(BytesWritten) /= Message.Count then

Text.Data(1..28) := "ERROR: Write of wrong length";

Text := TextIO.Concat(Text.Data(1..28),Integer(BytesWritten));

Text := TextIO.Concat(Text.Data(1..Text.Count),Integer(Message.Count));

TextIO.Put_Line(Text);

return False;

else

Text.Data(1..21) := "Transmit WriteFile OK";

Text := TextIO.Concat(Text.Data(1..21),Integer(BytesWritten));

TextIO.Put_Line(Text);

return True;

end if;

else

Text.Data(1..27) := "ERROR: Write to pipe failed";

Text := TextIO.Concat(Text.Data(1..21),to_Int(SenderData.List(Index).Sender));

TextIO.Put_Line(Text);

return False;

end if;

end;

-- Catch the IOException that is raised if the pipe is broken

-- or disconnected.

exception -- catch (IOException e)

when others =>

Text_IO.Put("ERROR: Exception attempting to Transmit for");

Int_IO.Put(Integer(SenderData.List(Index).ToId));

Text_IO.Put_Line(" ");

return False;

end Transmit;

begin -- launch "procedure"

SenderData.Count := 0;

end NamedPipe.Client;

NamedPipe-Server

(spec)

with Choose;

with ExecItf;

package NamedPipe.Server is

-- child package of Socket

pragma Elaborate_Body;

-- SocketServer Listener Data

type ListenerDataType

is record

ToId : Choose.ComponentIdsType; -- Name and Id of this component waiting to

ToName : Choose.ComponentNameType; -- receive the message & the callback

RecvCallback : Choose.ReceiveCallbackType; -- to return the message

FromName : Choose.ComponentNameType; -- Name and Id of the component from

FromId : Choose.ComponentIdsType; -- which message is to be received

ThreadId : Integer; -- Id of Receive thread

ComputerName : Choose.ComponentNameType; -- unneeded for Server

DelPipeName : NamedPipe.PipeNameType; -- pipename

PipeName : NamedPipe.PipeNameType; -- must be of the form \\.\pipe\pipename

Created : Boolean; -- True means pipe opened

Listener : ExecItf.HANDLE; -- Pipe handle

end record;

type ListenerListType

is array (1..Choose.ComponentIdsType'Last) of ListenerDataType;

type ListenerType

is record

Count : Choose.ComponentIdsType;

List : ListenerListType;

end record;

ListenerData

: ListenerType;

function Connect

( Index : in Integer

) return Boolean;

function Lookup

( Id : in Integer

) return Choose.ComponentIdsType;

function Request

-- Request a Client component pairing

( FromName : in String;

FromId : in Choose.ComponentIdsType;

ToName : in String;

ToId : in Choose.ComponentIdsType;

RecvCallback : in Choose.ReceiveCallbackType

) return Boolean;

end NamedPipe.Server;

(body)

with CStrings;

with Delivery;

with ExecItf;

with Interfaces.C;

with Itf;

with System;

with TextIO;

with Text_IO;

with Threads;

with Unchecked_Conversion;

package body NamedPipe.Server is

-- child package of Socket

package Int_IO is new Text_IO.Integer_IO( Integer );

procedure Callback

( Id : in Integer

);

procedure OpenReceivePipe

( Index : in Integer

);

function Request

-- Request a Server component pairing

( FromName : in String;

FromId : in Choose.ComponentIdsType;

ToName : in String;

ToId : in Choose.ComponentIdsType;

RecvCallback : in Choose.ReceiveCallbackType

) return Boolean is

Index : Integer;

MatchIndex : Delivery.LocationType;

Partner : Delivery.LocationType;

PipeSize : Integer;

TDigits : String(1..2);

Success : Boolean;

ThreadName : String(1..3);

ReceiveResult

-- Result of Install of Receive with Threads

: Threads.RegisterResult;

function to_Callback is new Unchecked_Conversion

( Source => System.Address,

Target => Threads.CallbackType );

function to_Ptr is new Unchecked_Conversion

( Source => System.Address,

Target => ExecItf.PCSTR );

function to_Int is new Unchecked_Conversion

( Source => ExecItf.PSOCKADDR,

Target => Integer );

function to_Int1 is new Unchecked_Conversion

( Source => System.Address,

Target => Integer );

use type Interfaces.C.Int;

use type Delivery.LocationType;

use type ExecItf.SOCKET;

use type Threads.InstallResult;

begin -- Request

if ListenerData.Count < Threads.MaxComponents then

Index := ListenerData.Count + 1;

ListenerData.Count := Index;

ListenerData.List(Index).FromName.Count := FromName'Length;

ListenerData.List(Index).FromName.Value(1..FromName'Length) := FromName;

ListenerData.List(Index).FromId := FromId;

ListenerData.List(Index).ToName.Count := ToName'Length;

ListenerData.List(Index).ToName.Value(1..ToName'Length) := ToName;

ListenerData.List(Index).ToId := ToId;

ListenerData.List(Index).RecvCallback := RecvCallback;

-- Find the partner in DeliveryTable. This is a validation as

-- well as that the invocating component is correct that the from

-- and to component ids and names match the table.

MatchIndex := Delivery.Lookup(Choose.Pipe, ToId, FromId);

-- Set the Computer Name and the pipes.

if MatchIndex > 0 then

Partner := Delivery.Partner( MatchIndex );

-- Fill in computer name and pipe

ListenerData.List(Index).ComputerName := Delivery.ComputerName(Partner);

ListenerData.List(Index).DelPipeName := Delivery.Pipe(Partner);

PipeSize := ListenerData.List(Index).DelPipeName.Count;

ListenerData.List(Index).PipeName.Value(1..9) := "\\.\pipe\";

for I in 1..PipeSize loop

ListenerData.List(Index).PipeName.Value(9+I) :=

ListenerData.List(Index).DelPipeName.Value(I);

end loop;

ListenerData.List(Index).PipeName.Count := PipeSize + 9;

ListenerData.List(Index).PipeName.Value(9+PipeSize+1) := ASCII.NUL;

Text_IO.Put( "MatchIndex " );

Int_IO.Put( Integer(MatchIndex) );

Text_IO.Put( " " );

Text_IO.Put( " ServerPipe " );

Text_IO.Put( ListenerData.List(Index).PipeName.Value(1..PipeSize+9) );

Text_IO.Put_Line( " " );

else

Text_IO.Put_Line( "ERROR: To-From not valid for Server" );

return False;

end if;

-- Create thread for receive.

ListenerData.List(Index).ThreadId := Threads.TableCount + 1;

-- index in table after Install

ThreadName(1..3) := "R00";

CStrings.IntegerToString( From => Index,

Size => 2,

CTerm => False,

Result => TDigits,

Success => Success );

ThreadName(2..3) := TDigits;

if ThreadName(2) = ' ' then

ThreadName(2) := '0';

end if;

ListenerData.List(Index).ThreadId := Index;

ReceiveResult := Threads.Install

( Name => ThreadName,

Index => ListenerData.List(Index).ThreadId,

Priority => Threads.HIGH, --NORMAL,

Callback => to_Callback(Callback'Address) );

declare

Text : Itf.V_80_String_Type;

begin

Text.Data(1..19) := "Install Recv Thread";

Text := TextIO.Concat(Text.Data(1..19), ThreadName(1..3));

if ReceiveResult.Status /= Threads.Valid then

Text := TextIO.Concat(Text.Data(1..Text.Count),"failed");

TextIO.Put_Line(Text);

return False;

else

Text := TextIO.Concat(Text.Data(1..Text.Count),"success");

Text := TextIO.Concat(Text.Data(1..Text.Count),to_Int1(Callback'Address));

TextIO.Put_Line(Text);

end if;

end;

-- Open the Receive Pipe

OpenReceivePipe( Index );

Text_IO.Put("ListenerData count ");

Int_IO.Put(ListenerData.Count);

Text_IO.Put(" ");

Int_IO.Put(fromId);

Text_IO.Put_Line(" ");

return True;

else

Text_IO.Put_Line( "ERROR: Too many Listeners" );

return False;

end if;

end Request;

function Lookup

( Id : in Integer

) return Choose.ComponentIdsType is

begin -- Lookup

for I in 1..ListenerData.Count loop

if ListenerData.List(I).ThreadId = Id then

return I;

end if;

end loop;

return 0;

end Lookup;

function to_Digit

( Number : in Integer

) return Character is

-- Convert number from 1 thru 9 to a alpha digit.

begin -- to_Digit

case Number is

when 1 => return '1';

when 2 => return '2';

when 3 => return '3';

when 4 => return '4';

when 5 => return '5';

when 6 => return '6';

when 7 => return '7';

when 8 => return '8';

when 9 => return '9';

when others =>

Text_IO.Put("ERROR: to_Digit for Number not 1 thru 0");

Int_IO.Put(Number);

Text_IO.Put_Line(" ");

return '0';

end case;

end to_Digit;

function Connect

( Index : in Integer

) return Boolean is

Handle

-- Handle of pipe

: ExecItf.HANDLE;

Status

-- True means server connected to client

: ExecItf.BOOL;

use type ExecItf.BOOL;

begin -- Connect

-- Wait for the client to connect. If it succeeds, the function returns

-- a nonzero value. If the function returns zero, GetLastError returns

-- ERROR_PIPE_CONNECTED.

Handle := ListenerData.List(Index).Listener;

Status := ExecItf.ConnectNamedPipe

( NamedPipe => Handle,

Overlapped => null );

if Status = 0 then -- FALSE

NamedPipe.Data.List(Index).Connected := False;

else -- TRUE

NamedPipe.Data.List(Index).Connected := True;

end if;

return NamedPipe.Data.List(Index).Connected;

end Connect;

-- Open the Receive Pipe

procedure OpenReceivePipe

( Index : in Integer

) is

Name : NamedPipe.PipeNameType;

Text : Itf.V_80_String_Type;

use type Interfaces.C.unsigned_long;

use type System.Address;

function AddrToLPSCSTR -- convert address to ExecItf pointer

is new Unchecked_Conversion( Source => System.Address,

Target => ExecItf.LPCSTR );

function to_Int is new Unchecked_Conversion

( Source => System.Address,

Target => Integer );

begin -- OpenReceivePipe

Name := ListenerData.List(Index).PipeName;

ListenerData.List(Index).Listener :=

ExecItf.CreateFile

( FileName => AddrToLPSCSTR(Name.Value'Address),

DesiredAccess => ExecItf.GENERIC_READ or ExecItf.GENERIC_WRITE,

ShareMode => 0, -- no sharing --ExecItf.FILE_SHARE_READ,

SecurityAttributes => null, -- default security attributes

CreationDisposition => ExecItf.OPEN_ALWAYS,

FlagsAndAttributes => 0, -- default attributes

TemplateFile => System.Null_Address -- no template

);

-- Note: The client and server processes in this example are intended

-- to run on the same computer, so the server name provided to the

-- NamedPipeClientStream object is ".". If the client and server

-- processes were on separate computers, "." would be replaced with

-- the network name of the computer that runs the server process.

-- This will be done when the PipeName.Value is created.

if ListenerData.List(Index).Listener = ExecItf.Invalid_Handle_Value then

null;

else

Text.Data(1..42) := "NamedPipe Server (Receive) Handle is Valid";

Text := TextIO.Concat(Text.Data(1..42),to_Int(ListenerData.List(Index).Listener));

TextIO.Put_Line(Text);

begin

NamedPipe.Data.List(Index).Connected := Connect( Index );

exception

when others => null;

end;

end if;

end OpenReceivePipe;

procedure Callback

( Id : in Integer

) is separate;

begin -- launch "procedure"

ListenerData.Count := 0;

end NamedPipe.Server;

(Callback)

separate (NamedPipe.Server)

-- Forever loop as initiated by Threads to Receive a message

procedure Callback

( Id : in Integer

) is

-- This procedure runs in the particular thread assigned to accept the

-- connection for a component and receive a message.

-- Each component has its own receive callback and invokes this function

-- for the common processing.

type Int_Ptr_Type is access ExecItf.INT;

use type Interfaces.C.int;

Index

-- Index for Component in Data Listener

: Choose.ComponentIdsType := Id;

Text : Itf.V_80_String_Type;

use type ExecItf.BOOL;

use type System.Address;

function to_Int is new Unchecked_Conversion

( Source => System.Address,

Target => Integer );

function to_Ptr is new Unchecked_Conversion

( Source => System.Address,

Target => ExecItf.PSTR );

function toLPDWORD -- convert address to Exec_Itf pointer

is new Unchecked_Conversion( Source => System.Address,

Target => ExecItf.LPDWORD );

function toLPVOID -- convert address to Exec_Itf pointer

is new Unchecked_Conversion( Source => System.Address,

Target => ExecItf.LPVOID );

begin -- Callback

Index := NamedPipe.Server.Lookup( Id );

Text.Data(1..14) := "Callback Index";

Text := TextIO.Concat(Text.Data(1..14), Id);

Text := TextIO.Concat(Text.Data(1..Text.Count), to_Int(Callback'address));

TextIO.Put_Line(Text);

if Index = 0 then

Text.Data(1..26) := "ERROR: Invalid Callback Id";

Text := TextIO.Concat(Text.Data(1..26), Id);

TextIO.Put_Line(Text);

loop

Delay(5.0);

end loop;

end if;

Forever:

loop

if NamedPipe.Server.ListenerData.List(Index).Listener = ExecItf.Invalid_Handle_Value then

NamedPipe.Server.OpenReceivePipe(Index => Index );

end if;

if NamedPipe.Server.ListenerData.List(Index).Listener /= ExecItf.Invalid_Handle_Value then

declare

BytesToRead : Interfaces.C.unsigned_long;

BytesRead : Integer;

Message : Itf.BytesType;

Status : ExecItf.BOOL;

use type ExecItf.BOOL;

begin

BytesToRead := Interfaces.C.unsigned_long(Message.Bytes'Last);

Status :=

ExecItf.ReadFile

( File => NamedPipe.Server.ListenerData.List(Index).Listener,

Buffer => toLPVOID(Message.Bytes'address), -- buffer to receive data

NumberOfBytesToRead => BytesToRead, -- size of the buffer

NumberOfBytesRead => toLPDWORD(BytesRead'address),

Overlapped => null ); -- not overlapped I/O

if BytesRead > Message.Bytes'Last then

Text.Data(1..19) := "Too many bytes read";

Text := TextIO.Concat(Text.Data(1..19),Integer(BytesRead));

TextIO.Put_Line(Text);

end if;

Message.Count := BytesRead;

if Status /= 0 then -- TRUE

if BytesRead = 0 then

Text_IO.Put_Line("ERROR: NamedPipe-Server Receive of 0 bytes");

elsif BytesRead > Itf.MessageSize then

Text_IO.Put_Line(

"ERROR: NamedPipe-Server Receive of more than MessageSize bytes");

exit; -- has to be from elsewhere

else -- Pass the message to its associated component

Message.Count := Integer(BytesRead);

NamedPipe.Server.ListenerData.List(Index).RecvCallback( Message => Message );

end if;

end if; -- Status /= 0

end;

end if;

Delay(0.03); -- allow thread to be switched

end loop Forever;

end Callback;

Note that the delivery pipe name (DelPipeName) is the short name – e.g.; 4to1. PipeName is the expansion of DelPipeName to include the prefix as required to form a valid Named Pipes pipe name.

Problem Encountered

It took me quite some time to figure out a problem that was occurring. When I attempted to create the receive message Callback as above (although not a separate file at that time) a data area address would be supplied rather than a code area address. Then a

Program received signal SIGSEGV, Segmentation fault.

exception would occur. It was after I started getting this fault that I found that the address was in the wrong area.

I couldn't figure out why. The threads for the components were OK and if I had the component also create a thread for the receive callback it would be OK. And I could then call a common Server Receive procedure that was almost the same as the above Callback procedure and have it execute.

After fussing about this for a number of days the thought occurred to me that I had named the Choose parameter to pass the received message back to the component as Callback as well. Such as

function Request

-- Request a Server component pairing

( Option : in OptionType := Server;

FromName : in String;

FromId : in ComponentIdsType;

ToName : in String;

ToId : in ComponentIdsType;

Callback : in ReceiveCallbackType

) return Boolean;

So as an experiment I changed to RecvCallback as in the Choose files above. This meant that the Request function of NamedPipe-Server had also referenced Callback as its similar parameter and was changed to RecvCallback. Although Callback wasn't used as a name to store the value, the compiler must have gotten confused. When I switched to RecvCallback as the name of the parameter in the Choose and Server Request functions, the problem was resolved. And I could return to the previous usage of having the receive Callback procedure in NamedPipe-Server for Threads to invoke.

Results

The routing using Named Pipes worked as well as when TCP Sockets were used.

The results from the running of App1:

in Component1 callback 2

Component1 to send to NewComponent 26 Message for NewComponent 1

ComponentThread 2 5 4266248 NewComponent

in NewComponent callback 6

Callback Index 1 4251526

NamedPipe Server (Receive) Handle is Valid 252

Client Transmit Index 1

Client Transmit connected for remote app 256 \\.\pipe\1to4

Client Transmit do Open

Transmit sending 26 bytes WriteFile

ERROR: Write to pipe 256

Message not sent to NewComComponent

NamedPipe Server (Receive) Handle is Valid 260

Component5 received a message: Message for Component5 3

Component1 to send to NewComponent 26 Message for NewComponent 2

Client Transmit Index 1

Transmit sending 26 bytes WriteFile

Transmit WriteFile OK 26

NewComponent received a message: Message for NewComponent 2

Component5 received a message: Message for Component5 4

Component5 to send to Component6 24 Message for Component6 1

Client Transmit Index 2

NewComponent to send to Component1 24 Message for Component1 1

Client Transmit Index 3

Client Transmit connected for remote app 264 \\.\pipe\5to6

Client Transmit do Open

Transmit sending 24 bytes WriteFile

ERROR: Write to pipe 264

Message not sent to Component6

Client Transmit connected for remote app 268 \\.\pipe\4to1

Client Transmit do Open

Transmit sending 24 bytes WriteFile

ERROR: Write to pipe 268

Message not sent to Component1

Component1 to send to NewComponent 26 Message for NewComponent 3 ß A

NamedPipe Server (Receive) Handle is Valid 272

Client Transmit Index 1

Transmit sending 26 bytes WriteFile

Transmit WriteFile OK 26

NewComponent received a message: Message for NewComponent 3 ß A

Component5 received a message: Message for Component5 5 ß B

NewComponent to send to Component1 24 Message for Component1 2 ß D

Client Transmit Index 3

Component5 to send to Component6 24 Message for Component6 2 ß C

Client Transmit Index 2

Transmit sending 24 bytes WriteFile

Transmit WriteFile OK 24

Component1 received a message: Message for Component1 2 ß D

Component1 to send to NewComponent 26 Message for NewComponent 4

Client Transmit Index 1

The results from the running of App2:

NamedPipe Server (Receive) Handle is Valid 208

Component6 to send to Component5 24 Message for Component5 5 ß B

Client Transmit Index 1

Transmit sending 24 bytes WriteFile

Transmit WriteFile OK 24

Component6 received a message: Message for Component6 2 ß C

Component6 to send to Component5 24 Message for Component5 6

Client Transmit Index 1

Transmit sending 24 bytes WriteFile

Transmit WriteFile OK 24

Component6 received a message: Message for Component6 3

Component6 to send to Component5 24 Message for Component5 7

Future

Besides being able to communicate via either TCP Sockets or Named Pipes it has just occurred to me that using other variations on the Delivery Framework of the more distant past that

1) Messages were delivered based upon what components desired to consume the message.

2) This allowed for multiple components to consume a particular message.

3) This also allowed for the broadcast of a message if all the components registered to consume it.

This new delivery method results in considerably less code than the previous framework. But it lacks these qualities. So a future activity (after allowing both Sockets and Pipes to be used) can be to see if messages can be delivered based on their message topic. Or at least broadcast to all the other components.

my head stuck in the computer

Wednesday, October 16, 2019

Kubernetes Using Named Pipes and TCP Sockets

Wild Goose Chase

Results

Problem With Using Named Pipes

Results having only one server to receive broadcast messages for a named pipe

Results having only a server to receive broadcast messages for each named pipe

Results using both named pipes and sockets to receive messages

Monday, July 29, 2019

Kubernets Using Named Pipes

About Me