Curator's Note: Here's a Redis how-to from back in 2011. Anyone who has heard of Redis has probably also heard of Resque, which is a lightweight queue'ing system. To the uninitiated it might seem strange, or maybe even impossible, to construct a queue'ing system using just a key-value store. In this article, I’m going to break down some of the primitives redis exposes that make building a queue'ing system over it trivial and show how Redis is so much more than just a key-value store. The problem Let’s say, you are a mathematician and have just come up with this super performant way of computing factors of numbers. You decide to write up the following sinatra service: def compute_factors(number) factors = crazy_performant_computation number end get "/compute_factors" do number, post_back_url = params[:number].to_i, params[:post_back_url] RestClient.post post_back_url, factors => compute_factors(number).to_json "OK" end You soon start seeing crazy traffic and realize, performant as your factor computation algorithm is, it’s not fast enough to keep up with the speed at which you are getting requests to your service. First pass at optimization by forking You realize that it’s going to be far more efficient to fork off a new Process or Thread and have that perform the computation and post back the result. So your code now changes to: get "/compute_factors" do number, post_back_url = params[:number].to_i, params[:post_back_url] Process.fork do RestClient.post post_back_url, factors => compute_factors(number).to_json end "OK" end While, this is great you soon realize that filling up the process table in your OS is not such a good idea. Capping process creation using a process pool It is exactly this problem that a process pool was meant to solve. The basic idea is that you would still like to perform your time-intensive task in the background, but would like to put a cap on the number of background processes you have running. There are some excellent libraries that solve this problem such as Delayed Job and Resque. However, being the hacker that you are, you decide to roll one yourself. There are however a bunch of issues that these libraries solve and you decide to pull a pen and paper and note them down to ensure that you are not missing anything: Cap how many workers you create You need to have a way to cap the number of background workers you create, that way you don’t have the same problem you were having before. Control worker creation and destruction You would like to be able to boot up and bring down your workers reasonably gracefully. Handle race conditions You realize, that spinning new processes means that you now have to ensure your code is concurrent-safe. Redis provides, some wonderful atomic operations out-of-the-box so this shouldn’t be too hard. Second pass using BRPOP Redis supports a couple of interesting data-structures including lists, sets and hashes. Redis lists have a command called RPOP which basically lets you pop an item off the tail of a list, in essence treating it like a queue. The RPOP command comes with a blocking variant of itself called BRPOP that blocks on the call to popping an element from the list. You can also specify a timeout for how long (in seconds) you would like to block on the call. def compute_factors(number) factors = crazy_performant_computation number end NUMBER_OF_WORKERS = (ENV['NUMBER_OF_WORKERS'] || 50).to_i NUMBER_OF_WORKERS.times do Process.fork do redis = Redis.new loop do val = redis.brpop "work_queue", 1 unless val puts "Process: #{Process.pid} is exiting" exit 0 end number, postback_url = Marshal.load val.last RestClient.post postback_url, factors => compute_factors(number).to_json end end end redis = Redis.new get "/compute_factors" do number, post_back_url = params[:number].to_i, params[:post_back_url] redis.lpush "work_queue", Marshal.dump([number, post_back_url]) "OK" end So you now have solved a bunch of problems in this new approach. We have a fixed number of workers running to handle our background processing – so now our process table getting filled is not subject to traffic conditions. Race conditions are handled for us by Redis, since BRPOP is atomic and guarantees no two workers will do duplicate work. And finally, workers destroy themselves if they break out of the brpop call due to their timeout being hit, in this case 1 second. So, that’s quite a slew of problems that have been solved for us by virtue of just using redis. We soon start, seeing a different problem though. As traffic in our site lags, workers seem to be dying off since their timeout is being hit. We’d really like to now have the workers block for a longer time than just 1 second, while also having the option to kill them off sooner if we need to. That way, they’ll not be hanging around for any longer than they have to. Gracefully shutting down workers Our mandate now is to shutdown our workers gracefully, using redis and little bit of UNIX signals magic (for examples of using signals in this area checkout Unicorn Is Unix and the Unicorn web-server. Our code now morphs to: def compute_factors(number) factors = crazy_performant_computation number end NUMBER_OF_WORKERS = (ENV['NUMBER_OF_WORKERS'] || 50).to_i NUMBER_OF_WORKERS.times do Process.fork do redis = Redis.new loop do val = redis.brpop "work_queue", 30 unless val puts "Process: #{Process.pid} is signing off due to timeout!" exit 0 end if val.last == "DIE!" puts "Process: #{Process.pid} has been asked to kill itself by parent" exit 0 end number, postback_url = Marshal.load val.last RestClient.post postback_url, factors => compute_factors(number).to_json end end end redis = Redis.new get "/compute_factors" do number, post_back_url = params[:number].to_i, params[:post_back_url] redis.lpush "work_queue", Marshal.dump([number, post_back_url]) "OK" end `echo #{Process.pid} > /tmp/factors.pid` puts "Parent process wrote PID to /tmp/factors.pid" trap('QUIT') do NUMBER_OF_WORKERS.times do redis.lpush "work_queue", "DIE!" end end We have now bumped up the timeout to 30 seconds and also have in place a way to bring down the workers near instantly. This is accomplished by the web-server trapping the QUIT signal and when it does, it pushes a “DIE!” message onto the redis “work_queue”. It pushes this message the same number of times as the NUMBER_OF_WORKERS. And since BRPOP is an atomic and concurrent-safe operation we are now supporting the bringing down of workers via redis. How cool is that! To gracefully shutdown the server and workers we just need to: kill -s QUIT `cat /tmp/factors.pid` Conclusion The next time you need to get some background job action going, stop yourself from just grabbing a library. Instead, toy around with redis lists a little. You’ll be surprised by how much you can accomplish with just straight redis primitives.

In order to receive JMS messages, Spring provides the concept of message listener containers. These are beans that can be tied to receive messages that arrive at certain destinations. This post will examine the different ways in which containers can be configured. A simple example is below where the DefaultMessageListenerContianer has been configured to watch one queue (the property jms.queue.name) and has a reference to a myMessageListener bean which implements the MessageListener interface (ie onMessage): This is all very well but means that the myMessageListener bean will have to handle the JMS Message object and process accordingly depending upon the type of javax.jms.Message and its payload. For example: if (message instanceof MapMessage) { // cast, get object, do something } An alternative is to use a MessageListenerAdapter. This class abstracts away the above processing and leaves your code to deal with just the message's payload. For example: The delegate is a reference to a myMessageReceiverDelegate bean which has one or more methods called processMessage. It does not need to implement the MessageListener interface. This method can be overload to handle different payload types. Spring behind the scenes will determine which gets called. For example: public void processMessage(final HashMap message) { // do something } public void processMessage(final String message) { // do something } For the given approach though, only one queue can be tied to the container. Another approach is to tie many listeners (therefore many queues) to the one container, The below Spring XML, using the jms namespace, shows how two listeners for different queues can be tied to one container: The myMessageReceiverDelegate bean is treated as an adapter delegate, therefore does not need to implement the MessageListener interface. Each listener can have a different delegate but for the above example, all messages arriving at the two queues are processed by the one receiver bean ie myMessageReceiverDelegate. If there is a need to check the message type and extract the payload, then the listener can use a class which implements the MessageListener interface (eg the myMessageListener bean used in the first example). The onMessage method will then be called when messages arrive at the specified destination:

Curator's Note: Check raw code snippets to see properly formatted tables. In my continued playing around with football data I wanted to write a cypher query against neo4j which would show me which teams had missed the most penalties this season and who missed them. I started off with a query that returned all the penalties that have been missed this season and the games those missed happened in: START player = node:players('name:*') MATCH player-[:missed_penalty_in]-game, player-[:played|subbed_on]-stats-[:in]-game, stats-[:for]-team, game-[:home_team]-home, game-[:away_team]-away RETURN player.name, team.name, home.name, away.name +-------------------------------------------------------------------------------------------------+ | player.name | team.name | home.name | away.name | +-------------------------------------------------------------------------------------------------+ | "Papiss Demba Cisse" | "Newcastle United" | "Newcastle United" | "Norwich City" | | "Wayne Rooney" | "Manchester United" | "Manchester United" | "Arsenal" | | "Mikel Arteta" | "Arsenal" | "Arsenal" | "Fulham" | | "David Silva" | "Manchester City" | "Manchester City" | "Southampton" | | "Frank Lampard" | "Chelsea" | "Manchester City" | "Chelsea" | | "Adel Taarabt" | "Queens Park Rangers" | "Queens Park Rangers" | "Norwich City" | | "Javier Hernández" | "Manchester United" | "Manchester United" | "Wigan Athletic" | | "Robin Van Persie" | "Manchester United" | "Southampton" | "Manchester United" | | "Jonathan Walters" | "Stoke City" | "Fulham" | "Stoke City" | | "Shane Long" | "West Bromwich Albion" | "West Bromwich Albion" | "Liverpool" | | "Steven Gerrard" | "Liverpool" | "Liverpool" | "West Bromwich Albion" | | "Lucas Piazon" | "Chelsea" | "Chelsea" | "Aston Villa" | +-------------------------------------------------------------------------------------------------+ 12 rows (there should actually be another penalty miss for Jonathan Walters against Chelsea but for some reason the data source has missed it off! I then grouped the penalty misses by team so that I’d have one row for each team and a collection showing the people who’d missed. We can use the COLLECT function to do the latter: START player = node:players('name:*') MATCH player-[:missed_penalty_in]-game, player-[:played|subbed_on]-stats-[:in]-game, stats-[:for]-team RETURN DISTINCT team.name, COLLECT(player.name) AS players I wanted to order the teams by the number of penalties they’d missed so Manchester United would be first in the table in this case and initially tried to order the results by a count of players: START player = node:players('name:*') MATCH player-[:missed_penalty_in]-game, player-[:played|subbed_on]-stats-[:in]-game, stats-[:for]-team RETURN DISTINCT team.name, COLLECT(player.name) AS players ORDER BY COUNT(player.name) which doesn’t actually compile: SyntaxException: Aggregation expressions must be listed in the RETURN clause to be used in ORDER BY I tried a few other variations such as the following: START player = node:players('name:*') MATCH player-[:missed_penalty_in]-game, player-[:played|subbed_on]-stats-[:in]-game, stats-[:for]-team RETURN DISTINCT team.name, COUNT(player.name) AS numberOfPlayers, COLLECT(player.name) AS players ORDER BY numberOfPlayers DESC which again doesn’t compile: SyntaxException: Aggregation expressions must be listed in the RETURN clause to be used in ORDER BY I eventually found a post by Andres where he explains that you need to split the query into two and make use of WITH if you want to make use of two aggregation expressions. I ended up with the following query which does the job: START player = node:players('name:*') MATCH player-[:missed_penalty_in]-game, player-[:played|subbed_on]-stats-[:in]-game, stats-[:for]-team WITH DISTINCT team, COLLECT(player.name) AS players MATCH player-[:missed_penalty_in]-game, player-[:played|subbed_on]-stats-[:in]-game, stats-[:for]-team WITH DISTINCT team, COUNT(player) AS numberOfPlayers, players RETURN team.name, players ORDER BY numberOfPlayers DESC +---------------------------------------------------------------------------------+ | team.name | players | +---------------------------------------------------------------------------------+ | "Manchester United" | ["Wayne Rooney","Javier Hernández","Robin Van Persie"] | | "Chelsea" | ["Frank Lampard","Lucas Piazon"] | | "Liverpool" | ["Steven Gerrard"] | | "Manchester City" | ["David Silva"] | | "Newcastle United" | ["Papiss Demba Cisse"] | | "Queens Park Rangers" | ["Adel Taarabt"] | | "Stoke City" | ["Jonathan Walters"] | | "Arsenal" | ["Mikel Arteta"] | | "West Bromwich Albion" | ["Shane Long"] | +---------------------------------------------------------------------------------+ 9 rows

We cannot imagine Internet service without TCP/IP. All Internet services we have developed and used at NHN are based on a solid basis, TCP/IP. Understanding how data is transferred via the network will help you to improve performance through tuning, troubleshooting, or introduction to a new technology. This article will describe the overall operation scheme of the network stack based on data flow and control flow in Linux OS and the hardware layer. Key Characteristics of TCP/IP How should I design a network protocol to transmit data quickly while keeping the data order without any data loss? TCP/IP has been designed with this consideration. The following are the key characteristics of TCP/IP required to understand the concept of the stack. TCP and IP Technically, since TCP and IP have different layer structures, it would be correct to describe them separately. However, here we will describe them as one. 1. Connection-oriented First, a connection is made between two endpoints (local and remote) and then data is transferred. Here, the "TCP connection identifier" is a combination of addresses of the two endpoints, having type. 2. Bidirectional Byte Stream Bidirectional data communication is made by using byte stream. 3. In-order Delivery A receiver receives data in the order of sending data from a sender. For that, the order of data is required. To mark the order, 32-bit integer data type is used. 4. Reliability through ACK When a sender did not receive ACK (acknowledgement) from a receiver after sending data to the receiver, the sender TCP re-sends the data to the receiver. Therefore, the sender TCP buffers unacknowledged data from the receiver. 5. Flow Control A sender sends as much data as a receiver can afford. A receiver sends the maximum number of bytes that it can receive (unused buffer size, receive window) to the sender. The sender sends as much data as the size of bytes that the receiver's receive window allows. 6. Congestion Control The congestion window is used separately from the receive window to prevent network congestion by limiting the volume of data flowing in the network. Like the receive window, the sender sends as much data as the size of bytes that the receiver's congestion window allows by using a variety of algorithms such as TCP Vegas, Westwood, BIC, and CUBIC. Different from flow control, congestion control is implemented by the sender only. Data Transmission As indicated by its name, a network stack has many layers. The following Figure 1 shows the layer types. Figure 1: Operation Process by Each Layer of TCP/IP Network Stack for Data Transmission. There are several layers and the layers are briefly classified into three areas: User area Kernel area Device area Tasks at the user area and the kernel area are performed by the CPU. The user area and the kernel area are called "host" to distinguish them from the device area. Here, the device is the Network Interface Card (NIC) that sends and receives packets. It is a more accurate term than the commonly used "LAN card". Let's take a look at the user area. First, the application creates data to send (the "User data" box in Figure 1) and then calls the write() system call to send the data. Assume that the socket (fd in Figure 1) has been already created. When the system call is called, the area is switched to the kernel area. POSIX-series operating systems including Linux and Unix expose the socket to the application by using a file descriptor. In the POSIX-series operating system, the socket is a kind of a file. The file layer executes a simple examination and calls the socket function by using the socket structure connected to the file structure. The kernel socket has two buffers. One is the send socket buffer for sending; And the other is the receive socket buffer for receiving. When the write system call is called, the data in the user area is copied to the kernel memory and then added to the end of the send socket buffer. This is to send data in order. In the Figure 1, the light-gray box refers to the data in the socket buffer. Then, TCP is called. There is the TCP Control Block (TCB) structure connected to the socket. The TCB includes data required for processing the TCP connection. Data in the TCB are connection state (LISTEN, ESTABLISHED, TIME_WAIT),receive window, congestion window, sequence number, resending timer, etc. If the current TCP state allows for data transmission, a new TCP segment (in other words, a packet) is created. If data transmission is impossible due to flow control or such a reason, the system call is ended here and then the mode is returned to the user mode (in other words, the control is passed to the application). There are two TCP segments as shown in Figure 2: TCP header; And payload. Figure 2: TCP Frame Structure (source). The payload includes the data saved in the unacknowledged send socket buffer. The maximum length of the payload is the maximum value among the receive window, congestion window, and maximum segment size (MSS). Then, TCP checksum is computed. In this checksum computation, pseudo header information (IP addresses, segment length, and protocol number) is included. One or more packets can be transmitted according to the TCP state. In fact, since the current network stack uses the checksum offload, the TCP checksum is computed by NIC, not by the kernel. However, we assume that the kernel computes the TCP checksum for convenience. The created TCP segment goes down to the IP layer. The IP layer adds an IP header to the TCP segment and performs IP routing. IP routing is a procedure of searching the next hop IP in order to go to the destination IP. After the IP layer has computed and added the IP header checksum, it sends the data to the Ethernet layer. The Ethernet layer searches for the MAC address of the next hop IP by using the Address Resolution Protocol (ARP). It then adds the Ethernet header to the packet. The host packet is completed by adding the Ethernet header. After IP routing is performed, the transmit interface (NIC) is known as the result of IP routing. The interface is used for transmitting a packet to the next hop IP and the IP. Therefore, the transmit NIC driver is called. At this time, if a packet capture program such as tcpdump or Wireshark is running, the kernel copies the packet data onto the memory buffer that the program uses. In that way, the receiving packet is directly captured on the driver. Generally, the traffic shaper function is implemented to run on this layer. The driver requests packet transmission according to the driver-NIC communication protocol defined by the NIC manufacturer. After receiving the packet transmission request, the NIC copies the packets from the main memory to its memory and then sends it to the network line. At this time, by complying with the Ethernet standard, it adds the IFG (Inter-Frame Gap), preamble, and CRC to the packet. The IFG and preamble are used to distinguish the start of the packet (as a networking term, framing), and the CRC is used to protect the data (the same purpose as TCP and IP checksum). Packet transmission is started based on the physical speed of the Ethernet and the condition of Ethernet flow control. It is like getting the floor and speaking in a conference room. When an NIC sends a packet, the NIC generates interrupts on the host CPU. Every interrupt has its own interrupt number and the OS searches an adequate driver to handle the interrupt by using the number. The driver registers a function to handle the interrupt (an interrupt handler) when the driver is started. The OS calls the interrupt handler and then the interrupt handler returns the transmitted packet to the OS. So far we have discussed the procedure of data transmission through the kernel and the device when an application performs write. However, without a direct write request from the application, the kernel can transmit a packet by directly calling TCP. For example, when an ACK is received and the receive window is expanded, the kernel creates a TCP segment including the data left in the socket buffer and sends the TCP segment to the receiver. Data Receiving Now, let's take a look at how data is received. Data receiving is a procedure for how the network stack handles a packet coming in. Figure 3 shows how the network stack handles a packet received. Figure 3: Operation Process by Each Layer of TCP/IP Network Stack for Handling Data Received. First, the NIC writes the packet onto its memory. It checks whether the packet is valid by performing the CRC check and then sends the packet to the memory buffer of the host. This buffer is a memory that has already been requested by the driver to the kernel and allocated for receiving packets. After the buffer has been allocated, the driver tells the memory address and size to the NIC. When there is no host memory buffer allocated by the driver even though the NIC receives a packet, the NIC may drop the packet. After sending the packet to the host memory buffer, the NIC sends an interrupt to the host OS. Then, the driver checks whether it can handle the new packet or not. So far, the driver-NIC communication protocol defined by the manufacturer is used. When the driver should send a packet to the upper layer, the packet must be wrapped in a packet structure that the OS uses for the OS to understand the packet. For example, sk_buff of Linux, mbuf of BSD-series kernel, and NET_BUFFER_LIST of Microsoft Windows are the packet structures of the corresponding OS. The driver sends the wrapped packets to the upper layer. The Ethernet layer checks whether the packet is valid and then de-multiplexes the upper protocol (network protocol). At this time, it uses the ethertype value of the Ethernet header. The IPv4 ethertype value is 0x0800. It removes the Ethernet header and then sends the packet to the IP layer. The IP layer also checks whether the packet is valid. In other words, it checks the IP header checksum. It logically determines whether it should perform IP routing and make the local system handle the packet, or send the packet to the other system. If the packet must be handled by the local system, the IP layer de-multiplexes the upper protocol (transport protocol) by referring to the proto value of the IP header. The TCP proto value is 6. It removes the IP header and then sends the packet to the TCP layer. Like the lower layer, the TCP layer checks whether the packet is valid. It also checks the TCP checksum. As mentioned before, since the current network stack uses the checksum offload, the TCP checksum is computed by NIC, not by the kernel. Then it searches the TCP control block where the packet is connected. At this time, of the packet is used as an identifier. After searching the connection, it performs the protocol to handle the packet. If it has received new data, it adds the data to the receive socket buffer. According to the TCP state, it can send a new TCP packet (for example, an ACK packet). Now TCP/IP receiving packet handling has completed. The size of the receive socket buffer is the TCP receive window. To a certain point, the TCP throughput increases when the receive window is large. In the past, the socket buffer size had been adjusted on the application or the OS configuration. The latest network stack has a function to adjust the receive socket buffer size, i.e., the receive window, automatically. When the application calls the read system call, the area is changed to the kernel area and the data in the socket buffer is copied to the memory in the user area. The copied data is removed from the socket buffer. And then the TCP is called. The TCP increases the receive window because there is new space in the socket buffer. And it sends a packet according to the protocol status. If no packet is transferred, the system call is terminated. Network Stack Development Direction The functions of network stack layers described so far are the most basic functions. The network stack in the early 1990s had few more functions than the functions described above. However, the latest network stack has many more functions and complexity as the network stack implementation structure gets higher. The latest network stack is classified by purpose as follows. Packet Processing Procedure Manipulation It is a function like Netfilter (firewall, NAT) and traffic control. By inserting the user-controllable code to the basic processing flow, the function can work differently according to the user configuration. Protocol Performance It aims to improve the throughput, latency, and stability that the TCP protocol can achieve within the given network environment. Various congestion control algorithms and additional TCP functions such as SACK are the typical examples. The protocol improvement will not be discussed here since it is out of the scope. Packet Processing Efficiency The packet processing efficiency aims to improve the maximum number of packets that can be processed per second by reducing the CPU cycle, memory usage, and memory accesses that one system consumes to process packets. There have been several attempts to reduce the latency in the system. The attempts include stack parallel processing, header prediction, zero-copy, single-copy, checksum offload, TSO, LRO, RSS, etc. Control Flow in the Stack Now, we will take a more detailed look at the internal flow of the Linux network stack. Like a subsystem which is not a network stack, a network stack basically runs as the event-driven way that reacts when the event occurs. Therefore, there is no separated thread to execute the stack. Figure 1 and Figure 3 showed the simplified diagrams of control flow. Figure 4 below illustrates more exact control flow. Figure 4: Control Flow in the Stack. At Flow (1) in Figure 4, an application calls a system call to execute (use) the TCP. For example, calls the read system call and the write system call and then executes TCP. However, there is no packet transmission. Flow (2) is same as Flow (1) if it requires packet transmission after executing TCP. It creates a packet and sends down the packet to the driver. A queue is in front of the driver. The packet comes into the queue first, and then the queue implementation structure decides the time to send the packet to the driver. This is queue discipline (qdisc) of Linux. The function of Linux traffic control is to manipulate the qdisc. The default qdisc is a simple First-In-First-Out (FIFO) queue. By using another qdisc, operators can achieve various effects such as artificial packet loss, packet delay, transmission rate limit, etc. At Flow (1) and Flow (2), the process thread of the application also executes the driver. Flow (3) shows the case in which the timer used by the TCP has expired. For example, when the TIME_WAITtimer has expired, the TCP is called to delete the connection. Like Flow (3), Flow (4) is the case in which the timer used by the TCP has expired and the TCP execution result packet should be transmitted. For example, when the retransmit timer has expired, the packet of which ACK has not been received is transmitted. Flow (3) and Flow (4) show the procedure of executing the timer softirq that has processed the timer interrupt. When the NIC driver receives an interrupt, it frees the transmitted packet. In most cases, execution of the driver is terminated here. Flow (5) is the case of packet accumulation in the transmit queue. The driver requests softirq and the softirq handler executes the transmit queue to send the accumulated packet to the driver. When the NIC driver receives an interrupt and finds a newly received packet, it requests softirq. The softirq that processes the received packet calls the driver and transmits the received packet to the upper layer. In Linux, processing the received packet as shown above is called New API (NAPI). It is similar to polling because the driver does not directly transmit the packet to the upper layer, but the upper layer directly gets the packet. The actual code is called NAPI poll or poll. Flow (6) shows the case that completes execution of TCP, and Flow (7) shows the case that requires additional packet transmission. All of Flow (5), (6), and (7) are executed by the softirq which has processed the NIC interrupt. How to Process Interrupt and Received Packet Interrupt processing is complex; however, you need to understand the performance issue related to processing of packets received. Figure 5 shows the procedure of processing an interrupt. Figure 5: Processing Interrupt, softirq, and Received Packet. Assume that the CPU 0 is executing an application program (user program). At this time, the NIC receives a packet and generates an interrupt for the CPU 0. Then the CPU executes the kernel interrupt (called irq) handler. This handler refers to the interrupt number and then calls the driver interrupt handler. The driver frees the packet transmitted and then calls the napi_schedule() function to process the received packet. This function requests the softirq (software interrupt). After execution of the driver interrupt handler has been terminated, the control is passed to the kernel handler. The kernel handler executes the interrupt handler for the softirq. After the interrupt context has been executed, the softirq context will be executed. The interrupt context and the softirq context are executed by an identical thread. However, they use different stacks. And, the interrupt context blocks hardware interrupts; however, the softirq context allows for hardware interrupts. The softirq handler that processes the received packet is the net_rx_action() function. This function calls thepoll() function of the driver. The poll() function calls the netif_receive_skb() function and then sends the received packets one by one to the upper layer. After processing the softirq, the application restarts execution from the stopped point in order to request a system call. Therefore, the CPU that has received the interrupt processes the received packets from the first to the last. In Linux, BSD, and Microsoft Windows, the processing procedure is basically the same on this wise. When you check the server CPU utilization, sometimes you can check that only one CPU executes the softirq hard among the server CPUs. The phenomenon occurs due to the way of processing received packets explained so far. To solve the problem, multi-queue NIC, RSS, and RPS have been developed. Data Structure The followings are some key data structures. Take a look at them and review the code. sk_buff structure First, there is the sk_buff structure or skb structure that means a packet. Figure 6 shows some of the sk_buffstructure. As the functions have been advanced, they get more complicated. However, the basic functions are very common that anyone can think. Figure 6: Packet Structure sk_buff. Including Packet Data and meta data The structure directly includes the packet data or refers to it by using a pointer. In Figure 6, some of the packets (from Ethernet to buffer) refer to using the data pointer and the additional data (frags) refer to the actual page. The necessary information such as header and payload length is saved in the meta data area. For example, inFigure 6, the mac_header, the network_header, and the transport_header have the corresponding pointer data that points the starting position of the Ethernet header, IP header and TCP header, respectively. This way makes TCP protocol processing easy. How to Add or Delete a Header The header is added or deleted as up and down each layer of the network stack. Pointers are used for more efficient processing. For example, to remove the Ethernet header, just increase the head pointer. How to Combine and Divide Packet The linked list is used for efficient execution of tasks such as adding or deleting packet payload data to the socket buffer, or packet chain. The next pointer and the prev pointer are used for this purpose. Quick Allocation and Free As a structure is allocated whenever creating a packet, the quick allocator is used. For example, if data is transmitted at the speed of 10-Gigabit Ethernet, more than one million packets per second must be created and deleted. TCP Control Block Second, there is a structure that represents the TCP connection. Previously, it was abstractly called a TCP control block. Linux uses tcp_sock for the structure. In Figure 7, you can see the relationship among the file, the socket, and the tcp_sock. Figure 7: TCP Connection Structure. When a system call has occurred, it searches the file in the file descriptor used by the application that has called the system call. For the Unix-series OS, the socket, the file and the device for general file system for storage are abstracted to a file. Therefore, the file structure includes the least information. For a socket, a separate socket structure saves the socket-related information and the file refers to the socket as a pointer. The socket refers to the tcp_sock again. The tcp_sock is classified into sock, inet_sock, etc to support various protocols except TCP. It may be considered as a kind of polymorphism. All status information used by the TCP protocol is saved in the tcp_sock. For example, the sequence number, receive window, congestion control, and retransmit timer are saved in the tcp_sock. The send socket buffer and the receive socket buffer are the sk_buff lists and they include the tcp_sock. The dst_entry, the IP routing result, is referred to in order to avoid too frequent routing. The dst_entry allows for easy search of the ARP result, i.e., the destination MAC address. The dst_entry is part of the routing table. The structure of the routing table is very complex that it will not be discussed in this document. The NIC to be used for packet transmission is searched by using the dst_entry. The NIC is expressed as the net_device structure. Therefore, by searching just the file, it is very easy to find all structures (from the file to the driver) required to process the TCP connection with the pointer. The size of the structures is the memory size used by one TCP connection. The memory size is a few KBs (excluding the packet data). As more functions have been added, the memory usage has been gradually increased. Finally, let's see the TCP connection lookup table. It is a hash table used to search the TCP connection where the received packet belongs. The hash value is calculated by using the input data of of the packet and the Jenkins hash algorithm. It is told that the hash function has been selected by considering defense against attacks to the hash table. Following Code: How to Transmit Data We will check the key tasks performed by the stack by following the actual Linux kernel source code. Here, we will observe two paths which are frequently used. First, this is a path used to transmit data when an application calls the write system call. SYSCALL_DEFINE3(write, unsigned int, fd, const char __user *, buf, ...) { struct file *file; [...] file = fget_light(fd, &fput_needed); [...] ===> ret = filp->f_op->aio_write(&kiocb, &iov, 1, kiocb.ki_pos); struct file_operations { [...] ssize_t (*aio_read) (struct kiocb *, const struct iovec *, ...) ssize_t (*aio_write) (struct kiocb *, const struct iovec *, ...) [...] }; static const struct file_operations socket_file_ops = { [...] .aio_read = sock_aio_read, .aio_write = sock_aio_write, [...] }; When the application calls the write system call, the kernel performs the write() function of the file layer. First, the actual file structure of the file descriptor fd is fetched. And then the aio_write is called. This is the function pointer. In the file structure, you will see the file_operations structure pointer. The structure is generally called function table and includes the function pointers such as aio_read and aio_write. The actual table for the socket is socket_file_ops. The aio_write function used by the socket is sock_aio_write. The function table is used for the purpose that is similar to the Java interface. It is generally used for the kernel to perform code abstraction or refactoring. static ssize_t sock_aio_write(struct kiocb *iocb, const struct iovec *iov, ..) { [...] struct socket *sock = file->private_data; [...] ===> return sock->ops->sendmsg(iocb, sock, msg, size); struct socket { [...] struct file *file; struct sock *sk; const struct proto_ops *ops; }; const struct proto_ops inet_stream_ops = { .family = PF_INET, [...] .connect = inet_stream_connect, .accept = inet_accept, .listen = inet_listen, .sendmsg = tcp_sendmsg, .recvmsg = inet_recvmsg, [...] }; struct proto_ops { [...] int (*connect) (struct socket *sock, ...) int (*accept) (struct socket *sock, ...) int (*listen) (struct socket *sock, int len); int (*sendmsg) (struct kiocb *iocb, struct socket *sock, ...) int (*recvmsg) (struct kiocb *iocb, struct socket *sock, ...) [...] }; The sock_aio_write() function gets the socket structure from the file and then calls sendmsg. It is also the function pointer. The socket structure includes the proto_ops function table. The proto_ops implemented by the IPv4 TCP is inet_stream_ops and the sendmsg is implemented by tcp_sendmsg. int tcp_sendmsg(struct kiocb *iocb, struct socket *sock, struct msghdr *msg, size_t size) { struct sock *sk = sock->sk; struct iovec *iov; struct tcp_sock *tp = tcp_sk(sk); struct sk_buff *skb; [...] mss_now = tcp_send_mss(sk, &size_goal, flags); /* Ok commence sending. */ iovlen = msg->msg_iovlen; iov = msg->msg_iov; copied = 0; [...] while (--iovlen >= 0) { int seglen = iov->iov_len; unsigned char __user *from = iov->iov_base; iov++; while (seglen > 0) { int copy = 0; int max = size_goal; [...] skb = sk_stream_alloc_skb(sk, select_size(sk, sg), sk->sk_allocation); if (!skb) goto wait_for_memory; /* * Check whether we can use HW checksum. */ if (sk->sk_route_caps & NETIF_F_ALL_CSUM) skb->ip_summed = CHECKSUM_PARTIAL; [...] skb_entail(sk, skb); [...] /* Where to copy to? */ if (skb_tailroom(skb) > 0) { /* We have some space in skb head. Superb! */ if (copy > skb_tailroom(skb)) copy = skb_tailroom(skb); if ((err = skb_add_data(skb, from, copy)) != 0) goto do_fault; [...] if (copied) tcp_push(sk, flags, mss_now, tp->nonagle); [...] } tcp_sengmsg gets tcp_sock (i.e.,TCP control block) from the socket and copies the data that the application has requested to transmit to the send socket buffer. When copying data to sk_buff, how many bytes will one sk_buff include? One sk_buff copies and includes MSS (tcp_send_mss) bytes to help the code that actually creates packets. Maximum Segment Size (MSS) stands for the maximum payload size that one TCP packet includes. By using TSO and GSO, one sk_buff can save more data than MSS. This will be discussed later, not in this document. The sk_stream_alloc_skb function creates a new sk_buff, and skb_entail adds the new sk_buff to the tail of the send_socket_buffer. The skb_add_data function copies the actual application data to the data buffer of thesk_buff. All the data is copied by repeating the procedure (creating an sk_buff and adding it to the send socket buffer) several times. Therefore, sk_buffs at the size of the MSS are in the send socket buffer as a list. Finally, the tcp_push is called to make the data which can be transmitted now as a packet, and the packet is sent. static inline void tcp_push(struct sock *sk, int flags, int mss_now, ...) [...] ===> static int tcp_write_xmit(struct sock *sk, unsigned int mss_now, ...) int nonagle, { struct tcp_sock *tp = tcp_sk(sk); struct sk_buff *skb; [...] while ((skb = tcp_send_head(sk))) { [...] cwnd_quota = tcp_cwnd_test(tp, skb); if (!cwnd_quota) break; if (unlikely(!tcp_snd_wnd_test(tp, skb, mss_now))) break; [...] if (unlikely(tcp_transmit_skb(sk, skb, 1, gfp))) break; /* Advance the send_head. This one is sent out. * This call will increment packets_out. */ tcp_event_new_data_sent(sk, skb); [...] The tcp_push function transmits as many of the sk_buffs in the send socket buffer as the TCP allows in sequence. First, the tcp_send_head is called to get the first sk_buff in the socket buffer and thetcp_cwnd_test and the tcp_snd_wnd_test are performed to check whether the congestion window and the receive window of the receiving TCP allow new packets to be transmitted. Then, the tcp_transmit_skb function is called to create a packet. static int tcp_transmit_skb(struct sock *sk, struct sk_buff *skb, int clone_it, gfp_t gfp_mask) { const struct inet_connection_sock *icsk = inet_csk(sk); struct inet_sock *inet; struct tcp_sock *tp; [...] if (likely(clone_it)) { if (unlikely(skb_cloned(skb))) skb = pskb_copy(skb, gfp_mask); else skb = skb_clone(skb, gfp_mask); if (unlikely(!skb)) return -ENOBUFS; } [...] skb_push(skb, tcp_header_size); skb_reset_transport_header(skb); skb_set_owner_w(skb, sk); /* Build TCP header and checksum it. */ th = tcp_hdr(skb); th->source = inet->inet_sport; th->dest = inet->inet_dport; th->seq = htonl(tcb->seq); th->ack_seq = htonl(tp->rcv_nxt); [...] icsk->icsk_af_ops->send_check(sk, skb); [...] err = icsk->icsk_af_ops->queue_xmit(skb); if (likely(err <= 0)) return err; tcp_enter_cwr(sk, 1); return net_xmit_eval(err); } tcp_transmit_skb creates the copy of the given sk_buff (pskb_copy). At this time, it does not copy the entire data of the application but the metadata. And then it calls skb_push to secure the header area and records the header field value. Send_check computes the TCP checksum. With the checksum offload, the payload data is not computed. Finally, queue_xmit is called to send the packet to the IP layer. Queue_xmit for IPv4 is implemented by the ip_queue_xmit function. int ip_queue_xmit(struct sk_buff *skb) [...] rt = (struct rtable *)__sk_dst_check(sk, 0); [...] /* OK, we know where to send it, allocate and build IP header. */ skb_push(skb, sizeof(struct iphdr) + (opt ? opt->optlen : 0)); skb_reset_network_header(skb); iph = ip_hdr(skb); *((__be16 *)iph) = htons((4 << 12) | (5 << 8) | (inet->tos & 0xff)); if (ip_dont_fragment(sk, &rt->dst) && !skb->local_df) iph->frag_off = htons(IP_DF); else iph->frag_off = 0; iph->ttl = ip_select_ttl(inet, &rt->dst); iph->protocol = sk->sk_protocol; iph->saddr = rt->rt_src; iph->daddr = rt->rt_dst; [...] res = ip_local_out(skb); [...] ===> int __ip_local_out(struct sk_buff *skb) [...] ip_send_check(iph); return nf_hook(NFPROTO_IPV4, NF_INET_LOCAL_OUT, skb, NULL, skb_dst(skb)->dev, dst_output); [...] ===> int ip_output(struct sk_buff *skb) { struct net_device *dev = skb_dst(skb)->dev; [...] skb->dev = dev; skb->protocol = htons(ETH_P_IP); return NF_HOOK_COND(NFPROTO_IPV4, NF_INET_POST_ROUTING, skb, NULL, dev, ip_finish_output, [...] ===> static int ip_finish_output(struct sk_buff *skb) [...] if (skb->len > ip_skb_dst_mtu(skb) && !skb_is_gso(skb)) return ip_fragment(skb, ip_finish_output2); else return ip_finish_output2(skb); The ip_queue_xmit function executes tasks required by the IP layers. __sk_dst_check checks whether the cached route is valid. If there is no cached route or the cached route is invalid, it performs IP routing. And then it calls skb_push to secure the IP header area and records the IP header field value. After that, as following the function call, ip_send_check computes the IP header checksum and calls the netfilter function. IP fragment is created when ip_finish_output function needs IP fragmentation. No fragmentation is generated when TCP is used. Therefore, ip_finish_output2 is called and it adds the Ethernet header. Finally, a packet is completed. int dev_queue_xmit(struct sk_buff *skb) [...] ===> static inline int __dev_xmit_skb(struct sk_buff *skb, struct Qdisc *q, ...) [...] if (...) { .... } else if ((q->flags & TCQ_F_CAN_BYPASS) && !qdisc_qlen(q) && qdisc_run_begin(q)) { [...] if (sch_direct_xmit(skb, q, dev, txq, root_lock)) { [...] ===> int sch_direct_xmit(struct sk_buff *skb, struct Qdisc *q, ...) [...] HARD_TX_LOCK(dev, txq, smp_processor_id()); if (!netif_tx_queue_frozen_or_stopped(txq)) ret = dev_hard_start_xmit(skb, dev, txq); HARD_TX_UNLOCK(dev, txq); [...] } int dev_hard_start_xmit(struct sk_buff *skb, struct net_device *dev, ...) [...] if (!list_empty(&ptype_all)) dev_queue_xmit_nit(skb, dev); [...] rc = ops->ndo_start_xmit(skb, dev); [...] } The completed packet is transmitted through the dev_queue_xmit function. First, the packet passes via the qdisc. If the default qdisc is used and the queue is empty, the sch_direct_xmit function is called to directly send down the packet to the driver, skipping the queue. Dev_hard_start_xmit function calls the actual driver. Before calling the driver, the device TX is locked first. This is to prevent several threads from accessing the device simultaneously. As the kernel locks the device TX, the driver transmission code does not need an additional lock. It is closely related to the parallel processing that will be discussed next time. Ndo_start_xmit function calls the driver code. Just before, you will see ptype_all and dev_queue_xmit_nit. The ptype_all is a list that includes the modules such as packet capture. If a capture program is running, the packet is copied by ptype_all to the separate program. Therefore, the packet that tcpdump shows is the packet transmitted to the driver. When checksum offload or TSO is used, the NIC manipulates the packet. So the tcpdump packet is different from the packet transmitted to the network line. After completing packet transmission, the driver interrupt handler returns the sk_buff. Following Code: How to Receive Data The general executed path is to receive a packet and then to add the data to the receive socket buffer. After executing the driver interrupt handler, follow the napi poll handle first. static void net_rx_action(struct softirq_action *h) { struct softnet_data *sd = &__get_cpu_var(softnet_data); unsigned long time_limit = jiffies + 2; int budget = netdev_budget; void *have; local_irq_disable(); while (!list_empty(&sd->poll_list)) { struct napi_struct *n; [...] n = list_first_entry(&sd->poll_list, struct napi_struct, poll_list); if (test_bit(NAPI_STATE_SCHED, &n->state)) { work = n->poll(n, weight); trace_napi_poll(n); } [...] } int netif_receive_skb(struct sk_buff *skb) [...] ===> static int __netif_receive_skb(struct sk_buff *skb) { struct packet_type *ptype, *pt_prev; [...] __be16 type; [...] list_for_each_entry_rcu(ptype, &ptype_all, list) { if (!ptype->dev || ptype->dev == skb->dev) { if (pt_prev) ret = deliver_skb(skb, pt_prev, orig_dev); pt_prev = ptype; } } [...] type = skb->protocol; list_for_each_entry_rcu(ptype, &ptype_base[ntohs(type) & PTYPE_HASH_MASK], list) { if (ptype->type == type && (ptype->dev == null_or_dev || ptype->dev == skb->dev || ptype->dev == orig_dev)) { if (pt_prev) ret = deliver_skb(skb, pt_prev, orig_dev); pt_prev = ptype; } } if (pt_prev) { ret = pt_prev->func(skb, skb->dev, pt_prev, orig_dev); static struct packet_type ip_packet_type __read_mostly = { .type = cpu_to_be16(ETH_P_IP), .func = ip_rcv, [...] }; As mentioned before, the net_rx_action function is the softirq handler that receives a packet. First, the driver that has requested the napi poll is retrieved from the poll_list and the poll handler of the driver is called. The driver wraps the received packet with sk_buff and then calls netif_receive_skb. When there is a module that requests all packets, the netif_receive_skb sends packets to the module. Like packet transmission, the packets are transmitted to the module registered to the ptype_all list. The packets are captured here. Then, the packets are transmitted to the upper layer based on the packet type. The Ethernet packet includes 2-byte ethertype field in the header. The value indicates the packet type. The driver records the value in sk_buff(skb->protocol). Each protocol has its own packet_type structure and registers the pointer of the structure to the ptype_base hash table. IPv4 uses ip_packet_type. The Type field value is the IPv4 ethertype (ETH_P_IP) value. Therefore, the IPv4 packet calls the ip_rcv function. int ip_rcv(struct sk_buff *skb, struct net_device *dev, ...) { struct iphdr *iph; u32 len; [...] iph = ip_hdr(skb); [...] if (iph->ihl < 5 || iph->version != 4) goto inhdr_error; if (!pskb_may_pull(skb, iph->ihl*4)) goto inhdr_error; iph = ip_hdr(skb); if (unlikely(ip_fast_csum((u8 *)iph, iph->ihl))) goto inhdr_error; len = ntohs(iph->tot_len); if (skb->len < len) { IP_INC_STATS_BH(dev_net(dev), IPSTATS_MIB_INTRUNCATEDPKTS); goto drop; } else if (len < (iph->ihl*4)) goto inhdr_error; [...] return NF_HOOK(NFPROTO_IPV4, NF_INET_PRE_ROUTING, skb, dev, NULL, ip_rcv_finish); [...] ===> int ip_local_deliver(struct sk_buff *skb) [...] if (ip_hdr(skb)->frag_off & htons(IP_MF | IP_OFFSET)) { if (ip_defrag(skb, IP_DEFRAG_LOCAL_DELIVER)) return 0; } return NF_HOOK(NFPROTO_IPV4, NF_INET_LOCAL_IN, skb, skb->dev, NULL, ip_local_deliver_finish); [...] ===> static int ip_local_deliver_finish(struct sk_buff *skb) [...] __skb_pull(skb, ip_hdrlen(skb)); [...] int protocol = ip_hdr(skb)->protocol; int hash, raw; const struct net_protocol *ipprot; [...] hash = protocol & (MAX_INET_PROTOS - 1); ipprot = rcu_dereference(inet_protos[hash]); if (ipprot != NULL) { [...] ret = ipprot->handler(skb); [...] ===> static const struct net_protocol tcp_protocol = { .handler = tcp_v4_rcv, [...] }; The ip_rcv function executes tasks required by the IP layers. It examines packets such as the length and header checksum. After passing through the netfilter code, it performs the ip_local_deliver function. If required, it assembles IP fragments. Then, it calls ip_local_deliver_finish through the netfilter code. Theip_local_deliver_finish function removes the IP header by using the __skb_pull and then searches the upper protocol whose value is identical to the IP header protocol value. Similar to the Ptype_base, each transport protocol registers its own net_protocol structure in inet_protos. IPv4 TCP uses tcp_protocol and callstcp_v4_rcv that has been registered as a handler. When packets come into the TCP layer, the packet processing flow varies depending on the TCP status and the packet type. Here, we will see the packet processing procedure when the expected next data packet has been received in the ESTABLISHED status of the TCP connection. This path is frequently executed by the server receiving data when there is no packet loss or out-of-order delivery. int tcp_v4_rcv(struct sk_buff *skb) { const struct iphdr *iph; struct tcphdr *th; struct sock *sk; [...] th = tcp_hdr(skb); if (th->doff < sizeof(struct tcphdr) / 4) goto bad_packet; if (!pskb_may_pull(skb, th->doff * 4)) goto discard_it; [...] th = tcp_hdr(skb); iph = ip_hdr(skb); TCP_SKB_CB(skb)->seq = ntohl(th->seq); TCP_SKB_CB(skb)->end_seq = (TCP_SKB_CB(skb)->seq + th->syn + th->fin + skb->len - th->doff * 4); TCP_SKB_CB(skb)->ack_seq = ntohl(th->ack_seq); TCP_SKB_CB(skb)->when = 0; TCP_SKB_CB(skb)->flags = iph->tos; TCP_SKB_CB(skb)->sacked = 0; sk = __inet_lookup_skb(&tcp_hashinfo, skb, th->source, th->dest); [...] ret = tcp_v4_do_rcv(sk, skb); First, the tcp_v4_rcv function validates the received packets. When the header size is larger than the data offset (th->doff < sizeof(struct tcphdr) / 4), it is the header error. And then __inet_lookup_skb is called to look for the connection where the packet belongs from the TCP connection hash table. From the sock structure found, all required structures such as tcp_sock and socket can be got. int tcp_v4_do_rcv(struct sock *sk, struct sk_buff *skb) [...] if (sk->sk_state == TCP_ESTABLISHED) { /* Fast path */ sock_rps_save_rxhash(sk, skb->rxhash); if (tcp_rcv_established(sk, skb, tcp_hdr(skb), skb->len)) { [...] ===> int tcp_rcv_established(struct sock *sk, struct sk_buff *skb, [...] /* * Header prediction. */ if ((tcp_flag_word(th) & TCP_HP_BITS) == tp->pred_flags && TCP_SKB_CB(skb)->seq == tp->rcv_nxt && !after(TCP_SKB_CB(skb)->ack_seq, tp->snd_nxt))) { [...] if ((int)skb->truesize > sk->sk_forward_alloc) goto step5; NET_INC_STATS_BH(sock_net(sk), LINUX_MIB_TCPHPHITS); /* Bulk data transfer: receiver */ __skb_pull(skb, tcp_header_len); __skb_queue_tail(&sk->sk_receive_queue, skb); skb_set_owner_r(skb, sk); tp->rcv_nxt = TCP_SKB_CB(skb)->end_seq; [...] if (!copied_early || tp->rcv_nxt != tp->rcv_wup) __tcp_ack_snd_check(sk, 0); [...] step5: if (th->ack && tcp_ack(sk, skb, FLAG_SLOWPATH) < 0) goto discard; tcp_rcv_rtt_measure_ts(sk, skb); /* Process urgent data. */ tcp_urg(sk, skb, th); /* step 7: process the segment text */ tcp_data_queue(sk, skb); tcp_data_snd_check(sk); tcp_ack_snd_check(sk); return 0; [...] } The actual protocol is executed from the tcp_v4_do_rcv function. If the TCP is in the ESTABLISHED status,tcp_rcv_esablished is called. Processing of the ESTABLISHED status is separately handled and optimized since it is the most common status. The tcp_rcv_established first executes the header prediction code. The header prediction is also quickly processed to detect in the common state. The common case here is that there is no data to transmit and the received data packet is the packet that must be received next time, i.e., the sequence number is the sequence number that the receiving TCP expects. In this case, the procedure is completed by adding the data to the socket buffer and then transmitting ACK. Go forward and you will see the sentence comparing truesize with sk_forward_alloc. It is to check whether there is any free space in the receive socket buffer to add new packet data. If there is, header prediction is "hit" (prediction succeeded). Then __skb_pull is called to remove the TCP header. After that, __skb_queue_tail is called to add the packet to the receive socket buffer. Finally, __tcp_ack_snd_check is called for transmitting ACK if necessary. In this way, packet processing is completed. If there is not enough free space, a slow path is executed. The tcp_data_queue function newly allocates the buffer space and adds the data packet to the socket buffer. At this time, the receive socket buffer size is automatically increased if possible. Different from the quick path, tcp_data_snd_check is called to transmit a new data packet if possible. Finally, tcp_ack_snd_check is called to create and transmit the ACK packet if necessary. The amount of code executed by the two paths is not much. This is accomplished by optimizing the common case. In other words, it means that the uncommon case will be processed significantly more slowly. The out-of-order delivery is one of the uncommon cases. How to Communicate between Driver and NIC Communication between a driver and the NIC is the bottom of the stack and most people do not care about it. However, the NIC is executing more and more tasks to solve the performance issue. Understanding the basic operation scheme will help you understand the additional technology. A driver and the NIC asynchronously communicate. First, a driver requests packet transmission (call) and the CPU performs another task without waiting for the response. And then the NIC sends packets and notifies the CPU of that, the driver returns the received packets (returns the result). Like packet transmission, packet receiving is asynchronous. First, a driver requests packet receiving and the CPU performs another task (call). Then, the NIC receives packets and notifies the CPU of that, and the driver processes the received packets received (returns the result). Therefore, a space to save the request and the response is necessary. In most cases, the NIC uses the ring structure. The ring is similar to the common queue structure. With the fixed number of entries, one entry saves one request data or one response data. The entries are sequentially used in turn. The name "ring" is generally used since the fixed entries are reused in turn. As following the packet transmission procedure shown in the following Figure 8, you will see how the ring is used. Figure 8: Driver-NIC Communication: How to Transmit Packet. The driver receives packets from the upper layer and creates the send descriptor that the NIC can understand. The send descriptor includes the packet size and the memory address by default. As the NIC needs the physical address to access the memory, the driver should change the virtual address of the packets to the physical address. Then, it adds the send descriptor to the TX ring (1). The TX ring is the send descriptor ring. Next, it notifies the NIC of the new request (2). The driver directly writes the data to a specific NIC memory address. In this way, Programmed I/O (PIO) is the data transmission method in which the CPU directly sends data to the device. The notified NIC gets the send descriptor of the TX ring from the host memory (3). Since the device directly accesses the memory without intervention of the CPU, the access is called Direct Memory Access (DMA). After getting the send descriptor, the NIC determines the packet address and the size and then gets the actual packets from the host memory (4). With the checksum offload, the NIC computes the checksum when the NIC gets the packet data from the memory. Therefore, overhead rarely occurs. The NIC sends packets (5) and then writes the number of packets that are sent to the host memory (6). Then, it sends an interrupt (7). The driver reads the number of packets that are sent and then returns the packets that have been sent so far. In the following Figure 9, you will see the procedure of receiving packets. Figure 9: Driver-NIC Communication: How to Receive Packets. First, the driver allocates the host memory buffer for receiving packets and then creates the receive descriptor. The receive descriptor includes the buffer size and the memory address by default. Like the send descriptor, it saves the physical address that the DMA uses in the receive descriptor. Then, it adds the receive descriptor to the RX ring (1). It is the receive request and the RX ring is the receive request ring. Through the PIO, the driver notifies that there is a new descriptor in the NIC (2). The NIC gets the new descriptor of the RX ring. And then it saves the size and location of the buffer included in the descriptor to the NIC memory (3). After the packets have been received (4), the NIC sends the packets to the host memory buffer (5). If the checksum offload function is existing, the NIC computes the checksum at this time. The actual size of received packets, the checksum result, and any other information are saved in the separate ring (the receive return ring) (6). The receive return ring saves the result of processing the receive request, i.e., the response. And then the NIC sends an interrupt (7). The driver gets packet information from the receive return ring and processes the received packets. If necessary, it allocates new memory buffer and repeats Step (1) and Step (2). To tune the stack, most people say that the ring and interrupt setting should be adjusted. When the TX ring is large, a lot of send requests can be made at once. When the RX ring is large, a lot of packet receives can be done at once. A large ring is useful for the workload that has a huge burst of packet transmission/receiving. In most cases, the NIC uses a timer to reduce the number of interrupts since the CPU may suffer from large overhead to process interrupts. To avoid flooding the host system with too many interrupts, interrupts are collected and sent regularly(interrupt coalescing) while sending and receiving the packets. Stack Buffer and Flow Control Flow control is executed in several stages in the stack. Figure 10 shows buffers used to transmit data. First, an application creates data and adds it to the send socket buffer. If there is no free space in the buffer, the system call is failed or the blocking occurs in the application thread. Therefore, the application data rate flowing into the kernel must be controlled by using the socket buffer size limit. Figure 10: Buffers Related to Packet Transmission. The TCP creates and sends packets to the driver through the transmit queue (qdisc). It is a typical FIFO queue type and the maximum length of the queue is the value of txqueuelen which can be checked by executing the ifconfig command. Generally, it is thousands of packets. The TX ring is between the driver and the NIC. As mentioned before, it is considered as a transmission request queue. If there is no free space in the queue, no transmission request is made and the packets are accumulated in the transmit queue. If too many packets are accumulated, packets are dropped. The NIC saves the packets to transmit in the internal buffer. The packet rate from this buffer is affected by the physical rate (ex: 1 Gb/s NIC cannot offer performance of 10 Gb/s). And with the Ethernet flow control, packet transmission is stopped if there is no free space in the receive NIC buffer. When the packet rate from the kernel is faster than the packet rate from the NIC, packets are accumulated in the buffer of the NIC. If there is no free space in the buffer, processing of transmission request from the TX ring is stopped. More and more requests are accumulated in the TX ring and finally there is no free space in the queue. The driver cannot make any transmission request and the packets are accumulated in the transmit queue. Like this, backpressure is sent from the bottom to the top through many buffers. Figure 11 shows the buffers that the receive packets are passing. The packets are saved in the receive buffer of the NIC. From the view of flow control, the RX ring between the driver and the NIC is considered as a packet buffer. The driver gets packets coming into the RX ring and then sends them to the upper layer. There is no buffer between the driver and the upper layer since the NIC driver that is used by the server system uses NAPI by default. Therefore, it can be considered as the upper layer directly gets packets from the RX ring. The payload data of packets is saved in the receive socket buffer. The application gets the data from the socket buffer later. Figure 11: Buffers Related to Packet Receiving. The driver that does not support NAPI saves packets in the backlog queue. Later, the NAPI handler gets packets. Therefore, the backlog queue can be considered as a buffer between the upper layer and the driver. If the packet processing rate of the kernel is slower than the packet flow rate into the NIC, the RX ring space is full. And the space of the buffer in the NIC is full, too. When the Ethernet flow control is used, the NIC sends a request to stop transmission to the transmission NIC or makes the packet drop. There is no packet drop due to lack of space in the receive socket buffer because the TCP supports end-to-end flow control. However, packet drop occurs due to lack of space in the socket buffer when the application rate is slow because the UDP does not support flow control. The sizes of the TX ring and the RX ring used by the driver in Figure 10 and Figure 11 are the sizes of the rings shown by the ethtool. For most workloads which regard throughput as important, it will be helpful to increase the ring size and the socket buffer size. Increasing the sizes reduces the possibility of failures caused by lack of space in the buffer while receiving and transmitting a lot of packets at a fast rate. Conclusion Initially, I planned to explain only the things that would be helpful for you to develop network programs, execute performance tests, and perform troubleshooting. In spite of my initial plan, the amount of description included in this document is not small. I hope this document will help you to develop network applications and monitor their performance. The TCP/IP protocol itself is very complicated and has many exceptions. However, you don't need to understand every line of TCP/IP-related code of the OS to understand performance and analyze the phenomena. Just understanding its context will be very helpful for you. With continuous advancement of system performance and implementation of the OS network stack, the latest server can offer 10-20 Gb/s TCP throughput without any problem. These days, there are too many technology types related to performance, such as TSO, LRO, RSS, GSO, GRO, UFO, XPS, IOAT, DDIO, and TOE, just like alphabet soup, to make us confused. In the next article, I will explain about the network stack from the performance perspective and discuss the problems and effects of this technology. By Hyeongyeop Kim, Senior Engineer at Performance Engineering Lab, NHN Corporation.

We have all been there. Looking at the poorly designed code while listening to the author’s explanations about how one should never sacrifice performance over design. And you just cannot convince the author to get rid of his 500-line methods because chaining method calls would destroy the performance. Well, it might have been true in 1996 or so. But since then JVM has evolved to be an amazing piece of software. One way to find out about it is to start looking more deeply into optimizations carried out by the virtual machine. The arsenal of techniques applied by the JVM is quite extensive, but lets look into one of them in more details. Namely method inlining . It is easiest to explain via the following sample: int sum(int a, int b, int c, int d) { return sum(sum(a, b),sum(c, d)); } int sum(int a, int b) { return a + b; } When this code is run, the JVM will figure out that it can replace it with a more effective, so called “inlined” code: int sum(int a, int b, int c, int d) { return a + b + c + d; } You have to pay attention that this optimization is done by the virtual machine and not by the compiler. It is not transparent at the first place why this decision was made. After all – if you look at the sample code above – why postpone optimization when compilation can produce more efficient bytecode? But considering also other not-so obvious cases, JVM is the best place to carry out the optimization: JVM is equipped with runtime data besides static analysis. During runtime JVM can make better decisions based on what methods are executed most often, what loads are redundant, when is it safe to use copy propagation, etc. JVM has got information about the underlying architecture – number of cores, heap size and configuration and can thus make the best selection based on this information. But let us see those assumptions in practice. I have created a small test application which uses several different ways to add together 1024 integers. A relatively reasonable one, where the implementation just iterates over the array containing 1024 integers and sums the result together. This implementation is available in InlineSummarizer.java . Recursion based divide-and-conquer approach. I take the original 1024 – element array and recursively divide it into halves – the first recursion depth thus gives me two 512-element arrays, the second depth has four 256-element arrays and so forth. In order to sum together all the 1024 elements I introduce 1023 additional method invocations. This implementation is attached as RecursiveSummarizer.java . Naive divide-and-conquer approach. This one also divides the original 1024-element array, but via calling additional instance methods on the separated halves – namely I nest sum512(), sum256(), sum128(), …, sum2() calls until I have summarized all the elements. As with recursion, I introduce 1023 additional method invocations in the source code . And I have a test class to run all those samples. The first results are from unoptimized code: As seen from the above, the inlined code is the fastest. And the ones where we have introduced 1023 additional method invocations are slower by ~25,000ns. But this image has to be interpreted with a caveat – it is a snapshot from the runs where JIT has not yet fully optimized the code. In my mid-2010 MB Pro it took between 200 and 3000 runs depending on the implementation. The more realistic results are below. I have ran all the summarizer implementations for more than 1,000,000 times and discarded the runs where JIT has not yet managed to perform it’s magic. We can see that even though inlined code still performed best, the iterative approach also flew at a decent speed. But recursion is notably different – when iterative approach close in with just 20% overhead, RecursiveSummarizer takes 340% of the time the inlined code needs to complete. Apparently this is something one should be aware of – when you use recursion, the JVM is helpless and cannot inline method calls. So be aware of this limitation when using recursion. Recursion aside – method overheads are close to being non-existent. Just 205 ns difference between having 1023 additional method invocations in your source code. Remember, those were nanoseconds (10^-9 s) over there that we used for measurement. So thanks to JIT we can safely neglect most of the overhead introduced by method invocations. The next time when your coworker is hiding his lousy design decisions behind the statement that popping through a call stack is not efficient, let him go through a small JIT crash course first. And if you wish to be well-equipped to block his future absurd statements, subscribe to either our RSS or Twitter feed and we are glad to provide you future case studies. Full disclosure: the inspiration for the test case used in this article was triggered by Tomasz Nurkiewicz blog post .

I was never sure what resources in JDBC must be explicitely closed and wasn’t able to find it anywhere explained. Finally my good colleague, Magne Mære, has explained it to me: In JDBC there are several kinds of resources that ideally should be closed after use. Even though every Statement and PreparedStatement is specified to be implicitly closed when the Connection object is closed, you can’t be guaranteed when (or if) this happens, especially if it’s used with connection pooling. You should explicitly close your Statement and PreparedStatement objects to be sure. ResultSet objects might also be an issue, but as they are guaranteed to be closed when the corresponding Statement/PreparedStatement object is closed, you can usually disregard it. Summary: Always close PreparedStatement/Statement and Connection. (Of course, with Java 7+ you’d use the try-with-resources idiom to make it happen automatically.) PS: I believe that the close() method on pooled connections doesn’t actually close them but just returns them to the pool. A request to my dear users: References to any good resources would be appreciate.

When working with MapReduce one of the challenges that is encountered early-on is determining how to make your third-part JAR’s available to the map and reduce tasks. One common approach is to create a fat jar, which is a JAR that contains your classes as well as your third-party classes (see this Cloudera blog post for more details). A more elegant solution is to take advantage of the libjars option in the hadoop jar command, also mentioned in the Cloudera post at a high level. Here I’ll go into detail on the three steps required to make this work. Add libjars to the options It can be confusing to know exactly where to put libjars when running the hadoop jar command. The following example shows the correct position of this option: $ export LIBJARS=/path/jar1,/path/jar2 $ hadoop jar my-example.jar com.example.MyTool -libjars ${LIBJARS} -mytoolopt value It’s worth noting in the above example that the JAR’s supplied as the value of the libjar option are comma-separated, and not separated by your O.S. path delimiter (which is how a Java classpath is delimited). You may think that you’re done, but often times this step alone may not be enough - read on for more details! Make sure your code is using GenericOptionsParser The Java class that’s being supplied to the hadoop jar command should use the GenericOptionsParser class to parse the options being supplied on the CLI. The easiest way to do that is demonstrated with the following code, which leverages the ToolRunner class to parse-out the options: public static void main(final String[] args) throws Exception { Configuration conf = new Configuration(); int res = ToolRunner.run(conf, new com.example.MyTool(), args); System.exit(res); } t is crucial that the configuration object being passed into the ToolRunner.run method is the same one that you’re using when setting-up your job. To guarantee this, your class should use the getConf() method defined in Configurable (and implemented in Configured) to access the configuration: public class SmallFilesMapReduce extends Configured implements Tool { public final int run(final String[] args) throws Exception { Job job = new Job(super.getConf()); ... job.waitForCompletion(true); return ...; } f you don’t leverage the Configuration object supplied to the ToolRunner.run method in your MapReduce driver code, then your job won’t be correctly configured and your third-party JAR’s won’t be copied to the Distributed Cache or loaded in the remote task JVM’s. It’s the ToolRunner.run method (actually it delegates the command parsing to GenericOptionsParser) which actually parses-out the libjars argument, and adds to the Configuration object a value for the tmpjarproperty. So a quick way to make sure that this step is working is to look at the job file for your MapReduce job (there’s a link when viewing the job details from the JobTracker), and make sure that the tmpjar configuration name exists with a value identical to the path that you specified in your command. You can also use the command-line to search for the libjars configuration in HDFS $ hadoop fs -cat /_logs/history/*.xml | grep tmpjars Use HADOOP_CLASSPATH to make your third-party JAR’s available on the client-side So far the first two steps tackled what you needed to do to to make your third-party JAR’s available to the remote map and reduce task JVM’s. But what hasn’t been covered so far is making these same JAR’s available to the client JVM, which is the JVM that’s created when you run the hadoop jar command. For this to happen, you should set the HADOOP_CLASSPATH environment variable to contain the O.S. path-delimited list of third-party JAR’s. Let’s extend the commands in the first step above with the addition of setting the HADOOP_CLASSPATH environment variable: $ export LIBJARS=/path/jar1,/path/jar2 $ export HADOOP_CLASSPATH=/path/jar1:/path/jar2 $ hadoop jar my-example.jar com.example.MyTool -libjars ${LIBJARS} -mytoolopt value Note that value for HADOOP_CLASSPATH uses a Unix path delimiter of :, so modify accordingly for your platform. And if you don’t like the copy-paste above you could modify that line to substitute the commas for semi-colons: $ export HADOOP_CLASSPATH=`echo ${LIBJARS} | sed s/,/:/g`

This is the second part of my text processing series. In this blog, we'll look into how text documents can be stored in a form that can be easily retrieved by a query. I'll used the popular open source Apache Lucene index for illustration. There are two main processing flow in the system ... Document indexing: Given a document, add it into the index Document retrieval: Given a query, retrieve the most relevant documents from the index. The following diagram illustrate how this is done in Lucene. Index Structure Both documents and query is represented as a bag of words. In Apache Lucene, "Document" is the basic unit for storage and retrieval. A "Document" contains multiple "Fields" (also call zones). Each "Field" contains multiple "Terms" (equivalent to words). To control how the document will be indexed across its containing fields, a Field can be declared in multiple ways to specified whether it should be analyzed (a pre-processing step during index), indexed (participate in the index) or stored (in case it needs to be returned in query result). Keyword (Not analyzed, Indexed, Stored) Unindexed (Not analyzed, Not indexed, Stored) Unstored (Analyzed, Indexed, Not stored) Text (Analyzed, Indexed, Stored) The inverted index is a core data structure of the storage. It is organized as an inverted manner from terms to the list of documents (which contain the term). The list (known as posting list) is ordered by a global ordering (typically by document id). To enable faster retrieval, the list is not just a single list but a hierarchy of skip lists. For simplicity, we ignore the skip list in subsequent discussion. This data structure is illustration below based on Lucene's implementation. It is stored on disk as segment files which will be brought to memory during the processing. The above diagram only shows the inverted index. The whole index contain an additional forward index as follows. Document indexing Document in its raw form is extracted from a data adaptor. (this can be making an Web API to retrieve some text output, or crawl a web page, or receiving an HTTP document upload). This can be done in a batch or online manner. When the index processing start, it parses each raw document and analyze its text content. The typical steps includes ... Tokenize the document (breakdown into words) Lowercase each word (to make it non-case-sensitive, but need to be careful with names or abbreviations) Remove stop words (take out high frequency words like "the", "a", but need to careful with phrases) Stemming (normalize different form of the same word, e.g. reduce "run", "running", "ran" into "run") Synonym handling. This can be done in two ways. Either expand the term to include its synonyms (ie: if the term is "huge", add "gigantic" and "big"), or reduce the term to a normalized synonym (ie: if the term is "gigantic" or "huge", change it to "big") At this point, the document is composed with multiple terms. doc = [term1, term2 ...]. Optionally, terms can be further combined into n-grams. After that we count the term frequency of this document. For example, in a bi-gram expansion, the document will become ... doc1 -> {term1: 5, term2: 8, term3: 4, term1_2: 3, term2_3:1} We may also compute a "static score" based on some measure of quality of the document. After that, we insert the document into the posting list (if it exist, otherwise create a new posting list) for each terms (all n-grams), this will create the inverted list structure as shown in previous diagram. There is a boost factor that can be set to the document or field. The boosting factor effectively multiply the term frequency which effectively affecting the importance of the document or field. Document can be added to the index in one of the following ways; inserted, modified and deleted. Typically the document will first added to the memory buffer, which is organized as an inverted index in RAM. When this is a document insertion, it goes through the normal indexing process (as I described above) to analyze the document and build an inverted list in RAM. When this is a document deletion (the client request only contains the doc id), it fetches the forward index to extract the document content, then goes through the normal indexing process to analyze the document and build the inverted list. But in this case the doc object in the inverted list is labeled as "deleted". When this is a document update (the client request contains the modified document), it is handled as a deletion followed by an insertion, which means the system first fetch the old document from the forward index to build an inverted list with nodes marked "deleted", and then build a new inverted list from the modified document. (e.g. If doc1 = "A B" is update to "A C", then the posting list will be {A:doc1(deleted) -> doc1, B:doc1(deleted), C:doc1}. After collapsing A, the posting list will be {A:doc1, B:doc1(deleted), C:doc1} As more and more document are inserted into the memory buffer, it will become full and will be flushed to a segment file on disk. In the background, when M segments files have been accumulated, Lucene merges them into bigger segment files. Notice that the size of segment files at each level is exponentially increased (M, M^2, M^3). This maintains the number of segment files that need to be search per query to be at the O(logN) complexity where N is the number of documents in the index. Lucene also provide an explicit "optimize" call that merges all the segment files into one. Here lets detail a bit on the merging process, since the posting list is already vertically ordered by terms and horizontally ordered by doc id, merging two segment files S1, S2 is basically as follows Walk the posting list from both S1 and S2 together in sorted term order. For those non-common terms (term that appears in one of S1 or S2 but not both), write out the posting list to a new segment S3. Until we find a common term T, we merge the corresponding posting list from these 2 segments. Since both list are sorted by doc id, we just walk down both posting list to write out the doc object to a new posting list. When both posting lists have the same doc (which is the case when the document is updated or deleted), we pick the latest doc based on time order. Finally, the doc frequency of each posting list (of the corresponding term) will be computed. Document retrieval Consider a document is a vector (each term as the separated dimension and the corresponding value is the tf-idf value) and the query is also a vector. The document retrieval problem can be defined as finding the top-k most similar document that match a query, where similarity is defined as the dot-product or cosine distance between the document vector and the query vector. tf-idf is a normalized frequency. TF (term frequency) represents how many time the term appears in the document (usually a compression function such as square root or logarithm is applied). IDF is the inverse of document frequency which is used to discount the significance if that term appears in many other documents. There are many variants of TF-IDF but generally it reflects the strength of association of the document (or query) with each term. Given a query Q containing terms [t1, t2], here is how we fetch the corresponding documents. A common approach is the "document at a time approach" where we traverse the posting list of t1, t2 concurrently (as opposed to the "term at a time" approach where we traverse the whole posting list of t1 before we start the posting list of t2). The traversal process is described as follows ... For each term t1, t2 in query, we identify all the corresponding posting lists. We walk each posting list concurrently to return a sequence of documents (ordered by doc id). Notice that each return document contains at least one term but can also also contain multiple terms. We compute the dynamic score which is dot product of the query to document vector. Notice that we typically don't concern the TF/IDF of the query (which is short and we don't care the frequency of each term). Therefore we can just compute the sum up all the TF score of the posting list that has a match term after dividing the IDF score (at the head of each posting list). Lucene also support query level boosting where a boost factor can be attached to the query terms. The boost factor will multiply the term frequency correspondingly. We also look up the static score which is purely based on the document (but not the query). The total score is a linear combination of static and dynamic score. Although the score we used in above calculation is based on computing the cosine distance between the query and document, we are not restricted to that. We can plug in any similarity function that make sense to the domain. (e.g. we can use machine learning to train a model to score the similarity between a query and a document). After we compute a total score, we insert the document into a heap data structure where the topK scored document is maintained. Here the whole posting list will be traversed. In case of the posting list is very long, the response time latency will be long. Is there a way that we don't have to traverse the whole list and still be able to find the approximate top K documents ? There are a couple strategies we can consider. Static Score Posting Order: Notice that the posting list is sorted based on a global order, this global ordering provide a monotonic increasing document id during the traversal that is important to support the "document at a time" traversal because it is impossible to visit the same document again. This global ordering, however, can be quite arbitrary and doesn't have to be the document id. So we can pick the order to be based on the static score (e.g. quality indicator of the document) which is global. The idea is that we traverse the posting list in decreasing magnitude of static score, so we are more likely to visit the document with the higher total score (static + dynamic score). Cut frequent terms: We do not traverse the posting list whose term has a low IDF value (ie: the term appears in many documents and therefore the posting list tends to be long). This way we avoid to traverse the long posting list. TopR list: For each posting list, we create an extra posting list which contains the top R documents who has the highest TF (term frequency) in the original list. When we perform the search, we perform our search in this topR list instead of the original posting list. Since we have multiple inverted index (in memory buffer as well as the segment files at different levels), we need to combine the result them. If termX appears in both segmentA and segmentB, then the fresher version will be picked. The fresher version is determine as follows; the segment with a lower level (smaller size) will be considered more fresh. If the two segment files are at the same level, then the one with a higher number is more fresh. On the other hand, the IDF value will be the sum of the corresponding IDF of each posting list in the segment file (the value will be slightly off if the same document has been updated, but such discrepancy is negligible). However, the processing of consolidating multiple segment files incur processing overhead in document retrieval. Lucene provide an explicit "optimize" call to merge all segment files into one single file so there is no need to look at multiple segment files during document retrieval. Distributed Index For large corpus (like the web documents), the index is typically distributed across multiple machines. There are two models of distribution: Term partitioning and Document partitioning. In document partitioning, documents are randomly spread across different partitions where the index is built. In term partitioning, the terms are spread across different partitions. We'll discuss document partitioning as it is more commonly used. Distributed index is provider by other technologies that is built on Lucene, such as ElasticSearch. A typical setting is as follows ... In this setting, machines are organized as columns and rows. Each column represent a partition of documents while each row represent a replica of the whole corpus. During the document indexing, first a row of the machines is randomly selected and will be allocated for building the index. When a new document crawled, a column machine from the selected row is randomly picked to host the document. The document will be sent to this machine where the index is build. The updated index will be later propagated to the other rows of replicas. During the document retrieval, first a row of replica machines is selected. The client query will then be broadcast to every column machine of the selected row. Each machine will perform the search in its local index and return the TopM elements to the query processor which will consolidate the results before sending back to client. Notice that K/P < M < K, where K is the TopK documents the client expects and P is the number of columns of machines. Notice that M is a parameter that need to be tuned. One caveat of this distributed index is that as the posting list is split horizontally across partitions, we lost the global view of the IDF value without which the machine is unable to calculate the TF-IDF score. There are two ways to mitigate that ... Do nothing: here we assume the document are evenly spread across different partitions so the local IDF represents a good ratio of the actual IDF. Extra round trip: In the first round, query is broadcasted to every column which returns its local IDF. The query processor will collected all IDF response and compute the sum of the IDF. In the second round, it broadcast the query along with the IDF sum to each column of machines, which will compute the local score based on the IDF sum.

The Producer Consumer pattern is an ideal way of separating work that needs to be done from the execution of that work. As you might guess from its name the Producer Consumer pattern contains two major components, which are usually linked by a queue. This means that the separation of the work that needs doing from the execution of that work is achieved by the Producer placing items of work on the queue for later processing instead of dealing with them the moment they are identified. The Consumer is then free to remove the work item from the queue for processing at any time in the future. This decoupling means that Producers don't care how each item of work will be processed, how many consumers will be processing it or how many other producers there are. It's a fire and forget world as far as they're concerned. Likewise consumers don't need to know where the work item came from, who put it in the queue, and how many other producers and consumers there are. All they need to do is to grab some work from the queue and process it. In the Java world, the Producer Consumer pattern is often based around some kind of blocking queue and there are several to choose from. These include ArrayBlockingQueue, LinkedBlockingQueue and PriorityBlockingQueue. Each have slightly different characteristics. The diagram above shows a simple implementation using a single pair of producer consumer objects, whilst the diagram below demonstrates how this can be expanded to include multiple producers and consumers. So, what about a practical scenario and some sample code? In the UK football is pretty popular (Soccer if you're reading this in the US) and every Saturday dozens of games are played throughout the land by a dedicated handful of professionals who sacrifice their afternoon in the pursuit of sporting excellence and large amounts of cash. A TV company sends a reporter to every game to feed live updates into a system and sent them back to the studio. On arriving at the studio the updates will be placed in a queue before being displayed on the screen by a Teletype. This scenario may have many producers, but only one or two consumers. This scenario is modelled by today's sample code using the class diagram shown below. The sample application, available on GitHub, is written as a Spring application because I want to separate the data from the code in a simple fashion, plus most readers of this blog already know about Spring, and it simplifies the scaffolding code that holds everything together. So far as Spring goes there are two Spring config files, matches.xml contains the match data whilst context.xml, shown below, contains a the Spring beans. The first task in designing any message based system is knowing the mechanism used to send your messages. In this case I've chosen a simple LinkedBlockingQueue and defined it in my Spring config. The second thing to do is to define what it is that you're sending and I'm sending in-play updates about the big game as demonstrated in the code below. public class Message implements Comparable { private final String name; private final long time; private final String matchTime; private final String messageText; public Message(String name, long time, String messageText, String matchTime) { this.name = name; this.time = time; this.messageText = messageText; this.matchTime = matchTime; } /** * @see java.lang.Comparable#compareTo(java.lang.Object) * * @return a negative integer, zero, or a positive integer as this object is * less than, equal to, or greater than the specified object */ @Override public int compareTo(Message compareTime) { int retVal = (int) (time - compareTime.time); return retVal; } @Override public String toString() { return matchTime + " - " + name + " - " + messageText; } public String getName() { return name; } public String getMessageText() { return messageText; } public long getTime() { return time; } public String getMatchTime() { return matchTime; } } This is a simple bean so there's not too much point in dwelling on it; however, during a match the will be a large number of these objects and they'll need organising and sorting, which is managed by the Match class below. public class Match { private final String name; private final List updates; public Match(String name, List matchInfo) { this.name = name; this.updates = new ArrayList(); createUpdateList(matchInfo); } private void createUpdateList(List matchInfo) { createMessageList(matchInfo); Collections.sort(updates); } private void createMessageList(List matchInfo) { for (String rawMessage : matchInfo) { final String timeString = getTime(rawMessage); final long time = parseTime(timeString); final String messageString = getMessage(rawMessage); Message message = new Message(name, time, messageString, timeString); updates.add(message); } } private String getTime(String rawMessage) { int index = rawMessage.indexOf(' '); String retVal = rawMessage.substring(0, index); return retVal; } /** * This may look weird, but the algorithm converts minutes to millis. eg 55:30 becomes * 55500mS */ private long parseTime(String timeString) { String[] split = timeString.split(":"); long minutes = (Long.valueOf(split[0]) * 1000); long seconds = (Long.valueOf(split[1])) * 1000 / 60; long time = minutes + seconds; return time; } private String getMessage(String rawMessage) { int index = rawMessage.indexOf(' '); String retVal = rawMessage.substring(index + 1); return retVal; } public String getName() { return name; } public List getUpdates() { return Collections.unmodifiableList(updates); } } This class takes a list of raw message strings as a constructor arg. Here's a snippet from matches.xml demonstrating the format of the messages: 95:21 Full time The referee blows his whistle to end the game. 94:06 Unfair challenge on Laurent Koscielny by Kenwyne Jones results in a free kick. Wojciech Szczesny takes the free kick. ...where the mm:ss component is the time of update. The Match class needs to load and sort the updates and this is simply achieved by creating a bunch of Message objects and then sorting them using Collections.sort() as as the Message class implements the Comparable interface. The next slice of code is the publisher and in this scenario it comes in the form of a MatchReporter. Each MatchReporter is allocated a Match and told about the queue via its constructor args. The MatchReporter is started by Spring calling its start() method as described in my blog on Three Spring Bean Lifecycle Techniques. Calling start() creates a new thread that allows the MatchReporter to check message times so that it can place them on the queue at the appropriate moment. public class MatchReporter implements Runnable { private final Match match; private final Queue queue; public MatchReporter(Match theBigMatch, Queue queue) { this.match = theBigMatch; this.queue = queue; } /** * Called by Spring after loading the context. Will "kick off" the match... */ public void start() { String name = match.getName(); Thread thread = new Thread(this, name); thread.start(); } /** * The main run loop */ @Override public void run() { long now = System.currentTimeMillis(); List matchUpdates = match.getUpdates(); for (Message message : matchUpdates) { delayUntilNextUpdate(now, message.getTime()); queue.add(message); } } private void delayUntilNextUpdate(long now, long messageTime) { while (System.currentTimeMillis() < now + messageTime) { try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } } } Note that in this example I've converted minutes and seconds in to milliseconds so that the app runs in a reasonable amount of time. For example, 55:30 becomes 55500mS. Having written the publisher code, the next thing to do is to sort out the consumer. In this example, the match updates are consumed by the Teletype (For those of you who don't know a teletype is take a look at Google. In the old days a TV camera used to be focused on a Teletype to bring viewers the latest scores). The Teletype has the job of reading any messages on the queue and displaying them on the screen. public class Teletype implements Runnable { private final BlockingQueue queue; private final PrintHead printHead; public Teletype(PrintHead printHead, BlockingQueue queue) { this.queue = queue; this.printHead = printHead; } public void start() { Thread thread = new Thread(this, "Studio Teletype"); thread.start(); } @Override public void run() { while (true) { try { Message message = queue.take(); printHead.print(message.toString()); } catch (InterruptedException e) { // TODO add some real error handling here printHead.print("Teletype error - try switching it off and on."); } } } public void destroy() { // Blank TODO... } } The Teletype code is much the sames as the MatchReporter code. Again I'm using Spring to start the ball rolling via Teletype's start() method and again it creates a new thread to carry out its work. The difference here is that queue.take() is a blocking call meaning that program execution will suspend at this point until there's at least one update on the queue. When a message is available queue.take() will whip it from the queue and it'll then be displayed on the screen using the PrintHead class. Once the message has been printed the run loop goes back to queue.take() for the next message where it'll block again until one appears on the queue. The code for this sample is available on Github and the eagle-eyed will have spotted that the tests in the TeletypeTest class have been disabled. This is because, although the Teletype code works it contains a couple of neat little flaws in that there's no way of shutting it down and that it's not particularly testable. As a developer you may not care too much about not being able to close your app down, but as an ops guy you do as starting and stopping stuff is pretty fundamental. In terms of the producer consumer patern there a few approaches you could take to rectify these problems, but more on that later... The code for this sample is available on GitHub.

This post comes from Ignacio Nin at the MySQL Performance Blog. Lately we’ve had many reports of the RPM packages for CentOS 5 (mostly) and CentOS 6 having issues when installing different combinations of our products, particularly with Percona Toolkit. Examples of bugs related to these issues are lp:1031427 and lp:1051874. These problems arise when trying to install a package from the distribution that is linked against the version of libmysqlclient.so shipped by the distribution (libmysqlclient.so.15 for CentOS 5/libmysqlclient.so.16 for CentOS 6) and a version of Percona Server that depends on another version of libmysqlclient.so, usually more recent. Bug lp:1031427 is an example of this, and shows how the packages would conflict when trying to install libmysqlclient.so. For example, when installing php-mysql alongside PS 5.5 in CentOS 6: # yum -q install Percona-Server-server-55 php-mysql Installing: Percona-Server-server-55 x86_64 5.5.29-rel29.4.401.rhel6 percona 15 M php-mysql x86_64 5.3.3-14.el6_3 updates 79 k Installing for dependencies: Percona-Server-client-55 x86_64 5.5.29-rel29.4.401.rhel6 percona 7.0 M Percona-Server-shared-51 x86_64 5.1.67-rel14.3.506.rhel6 percona 2.8 M Percona-Server-shared-55 x86_64 5.5.29-rel29.4.401.rhel6 percona 787 k Transaction Summary ===================================================================================================================================================== Install 5 Package(s) Is this ok [y/N]: y Transaction Check Error: file /usr/lib64/libmysqlclient.so conflicts between attempted installs of Percona-Server-shared-51-5.1.67-rel14.3.506.rhel6.x86_64 and Percona-Server-shared-55-5.5.29-rel29.4.401.rhel6.x86_64 file /usr/lib64/libmysqlclient_r.so conflicts between attempted installs of Percona-Server-shared-51-5.1.67-rel14.3.506.rhel6.x86_64 and Percona-Server-shared-55-5.5.29-rel29.4.401.rhel6.x86_64 The traditional solution for this situation was to provide a special package, Percona-Server-shared-compat (modeled after upstream’s MySQL-shared-compat) which would contain ALL versions of libmysqlclient.so.* together and wouldn’t conflict. Probably some of you are familiar with this approach. # yum -q install Percona-Server-server-55 Percona-Server-shared-compat php-mysql Installing: Percona-Server-server-55 x86_64 5.5.29-rel29.4.401.rhel6 percona 15 M Percona-Server-shared-compat x86_64 5.5.29-rel29.4.401.rhel6 percona 3.4 M php-mysql x86_64 5.3.3-14.el6_3 updates 79 k Installing for dependencies: Percona-Server-client-55 x86_64 5.5.29-rel29.4.401.rhel6 percona 7.0 M Percona-Server-shared-55 x86_64 5.5.29-rel29.4.401.rhel6 percona 787 k Transaction Summary ===================================================================================================================================================== Install 5 Package(s) Notice how PS-shared-compat installs along the -shared package, providing the older libmysqlclient.so.16 required by php-mysql. However, this has proved non-intuitive and problematic, since the shared-compat package wouldn’t get selected unless explicitely installed — and many of our users would rather have it “just work” without requiring additional knowledge of what the particular workaround was, etc.. We’re now trying a solution in which our -shared packages won’t conflict anymore at libmysqlclient.so, so we are able to install them side-by-side, modelled after the mysql-libs packages provided by CentOS/Redhat. So even if the user wants to install PS 5.5 alongside packages that depend on 5.1/5.0, the -shared packages will work together. For example installing 5.5 and postfix in CentOS: # yum -q install Percona-Server-server-55 postfix Installing: Percona-Server-server-55 x86_64 5.5.29-rel29.4.402.rhel5 percona-testing 19 M postfix x86_64 2:2.3.3-6.el5 base 3.8 M Installing for dependencies: Percona-SQL-shared-50 x86_64 5.0.92-b23.89.rhel5 percona-testing 1.8 M Percona-Server-client-55 x86_64 5.5.29-rel29.4.402.rhel5 percona-testing 9.1 M Percona-Server-shared-55 x86_64 5.5.29-rel29.4.402.rhel5 percona-testing 993 k … and this will install without problems. Additionally, this has the advantage of allowing an upgrade from 5.1 to 5.5 without uninstalling any software that depended on the old version. # rpm -qa | grep ^Percona Percona-Server-client-51-5.1.67-rel14.3.507.rhel6.x86_64 Percona-Server-shared-51-5.1.67-rel14.3.507.rhel6.x86_64 Percona-Server-server-51-5.1.67-rel14.3.507.rhel6.x86_64 In this case only Percona-Server-client-51 and Percona-Server-server-51 need be removed, allowing any package that depends on Percona-Server-shared-51 (providing libmysqlclient.so.16) to remain installed. After the server and client packages are uninstalled, you can install PS 5.5 without conflict. The current package candidates for versions 5.0.92 (which required an update), 5.1.67-14.3 and 5.5.29-29.4 can be tested from the percona-testing repository. We encourage you to try these out and send us your feedback and/or file any bugs you find. Installation instructions for Percona Testing repositories. We’re aiming to include these fixes in our next releases of 5.1 and 5.5. Percona Toolkit users in particular will enjoy this update since it’ll mean no more trouble when installing it from repository!

The post titled Installing Packages described the basics of package installation with R. The process is wonderfully simple when everything goes well. But it can be maddening when it does not. Error messages give a hint as to what went wrong but do not necessarily tell you how to resolve the problem. This post will collect some of the error messages we’ve encountered while installing R packages and describe the reasons for the error and the workarounds we’ve found. 1) Older version of R Warning message: In install.packages(c("sp")) : package ‘sp’ is not available This is the message that you get when the CRAN package you’re interested in requires a more recent version of R than you have. Remember, the default behavior ofinstall.packages() is to grab the latest version of a package. In this case you have to poke around in the “Old sources” link on the CRAN page for that package and use trial-and-error to find an older version of the package that will work with your version of R. You should start by determining what version of R you have: 1 2 $R--version Rversion2.8.1(2008-12-22) This version of R was released at the end of 2008 and any version of the “sp” package released in 2008 should work. At least some of the 2009 releases should also work. Perusing the sp archive, we might try installing version 0.9-37, the last of the 0.9-3x series which was released in May of 2009: 1 2 3 4 $wget http://cran.r-project.org/src/contrib/Archive/sp/sp_0.9-37.tar.gz $sudo CMD INSTALL sp_0.9-37.tar.gz ... $# Success! 2) Unable to execute files in /tmp directory ERROR: 'configure' exists but is not executable -- see the 'R Installation and Administration Manual' By default, R uses the /tmp directory to install packages. On security conscious machines, the /tmp directory is often marked as “noexec” in the /etc/fstab file. This means that no file under /tmp can ever be executed. Packages that require compilation or that have self-inflating data will fail with the error above. One such package isRJSONIO. The solution is to set the TMPDIR environment variable which R will use as the compilation directory. For csh shell: 1 2 $mkdir~/tmp $setenv TMPDIR~/tmp And for bash: 1 2 $mkdir~/tmp $export TMPDIR=~/tmp Other problems Please leave comments describing other package installation problems and solutions you’ve encountered to help us build a more complete listing.

RabbitMQ Simulator from Alvaro Videla on Vimeo. A demo of the new version of the RabbitMQ simulator with support for several exchange types, importing and exporting configuration and more. he goal of this app is to use it as a teaching tool for tutorials, presentations & more. Also it could serve as a topology designer for messaging applications since it supports exporting the design to RabbitMQ. Video Sections: - Intro - Direct Exchange - Fanout Exchange - Topic Exchange - Import/Export Configuration - Advanced Mode

When I want to insert figures generated in R into a LaTeX document, it looks better if I first remove the white space around the figure. Unfortunately, R does not make this easy as the graphs are generated to look good on a screen, not in a document. There are two things that can be done to fix this problem. First, you can reduce the white space generated by R. I use the following function when saving figures in R. savepdf <- function(file, width=16, height=10) { fname <- paste("figures/",file,".pdf",sep="") pdf(fname, width=width/2.54, height=height/2.54, pointsize=10) par(mgp=c(2.2,0.45,0), tcl=-0.4, mar=c(3.3,3.6,1.1,1.1)) } The width and height are in centimetres. The ratio is about right for a beamer presentation, and also to fit two figures on an A4 page. Then I use the commands savepdf("filename") # Plotting commands here dev.off() That will generate a pdf figure of about the right size and shape for a document, and with narrow margins of white space, and save it in my figures sub-directory. The second trick is to trim the pdf files so there is no white space left. On a unix system, this is easily achieved as follows. pdfcrop filename.pdf filename.pdf There are probably windows and mac versions of the same, but I haven’t used them. Adobe Acrobat will also crop pdfs, but not from the command line as far as I know. To apply pdfcrop to every file in a directory (using unix), save the following to a file called cropall.sh: #!/bin/bash for FILE in ./*.pdf; do pdfcrop "${FILE}" "${FILE}" done Make the file executable and run it. In my post on Makefiles, I explain how to include pdfcrop within a Makefile. If you just use pdfcrop without first reducing the white space in R, the proportions come out a little odd. So I tend to use both approaches together.

Please find the following sample which demonstrates the Nested iterator. The following is the source of the configuration. 15000 $1 Use the following request with soapui in order to test the above sample. SUN9 SUN10 From the second iterate mediator, messages retrieved from the first iterator are again iterated into smaller messages as below SUN10

There are times when you want to retrieve the ID of the last inserted record when using Entity Framework. For example: Employee emp = new Employee(); emp.ID = -1; emp.Name = "Senthil Kumar B"; emp.Expertise = "ASP.NET MVC" EmployeeContext context = new EmployeeContext(); context.AddObject(emp); context.SaveChanges(); In the above example , if i need to retrieve the ID of the employee that was inserted , all that i need to do is use the emp.ID property once the data is saved as shown below. Employee emp = new Employee(); emp.ID = -1; emp.Name = "Senthil Kumar B"; emp.Expertise = "ASP.NET MVC" EmployeeContext context = new EmployeeContext(); context.AddObject(emp); context.SaveChanges(); int empID = emp.ID;

JDK6 update 7 and onward include a tool called VisualVM. VisualVM is a visual tool with monitoring and profiling capabilities for the JVM. With VisualVM you can: Monitor heap usage Monitor CPU usage Monitor Threads Initiate garbage collections Profile CPU and memory And more… Although VisualVM is distributed with the Oracle JDK, it can also be used to monitor IBM JVM’s. VisualVM is not able to connect to the IBM JVM locally. JMX must be used instead. To enable JMX monitoring on the IBM JVM open the WebSphere administrative console and: navigate to: Server -> Server Types -> WebSphere application servers ->[SERVER_NAME] Expand Java and Process Management and click Process definition Click Java Virtual Machine In the Generic JVM arguments field append the following properties: -Djavax.management.builder.initial= -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dcom.sun.management.jmxremote.port=1099 Restart the server To start monitoring the JVM with VisualVM start VisualVM by navigating to [JDK_HOME]\bin and start jvisualvm.exe (please note that when VisualVM is downloaded as a separate package the executable is called visualvm.exe instead). In VisualVM click File -> Add JMX Connection. Specify localhost:1099 in the connection field and click OK. If everything went OK, you should see the localhost:1099 connection under the Local node in the tree on the left. Double-click this node to start monitoring. See the following screenshot for an example: When using a JMX connection to monitor the JVM please be aware that not all functionality can be used compared to monitoring a local JVM. Profiling memory is for example not possible. The above configuration was tested on: Windows 7 64-bit IBM JDK1.6 64 bit WebSphere Application Server version 7.0 Note: Before JDK6 update 7, VisualVM can also be downloaded separately from http://visualvm.java.net/download.html Note: To actually test if port 1099 is listening for connections use (on Windows) the netstat –a command and check wether the port is present and listening.

In a standard enterprise application which has a database and graphical UI (web or desktop), there are some typical layers which constitute the application. In this article we will mention those layers and give some instructions about them. The general diagram is as below: There are some important properties about that diagram: There are vertical and horizontal layers. Vertical layers may be thought as general application service libraries which can paralelly work on all horizontal layers and independent from each other. However, a horizontal layer's subject of interest is only its neighbour (top and bottom) horizontal layers, and they work sequentially from top-to-bottom and bottom-to-top. Lastly, user can only interact with topmost horizontal layer. Horizontal layers: Presentation: Every enterprise application has a UI, in fact graphical UI. This UI can be web or desktop based, which doesn't matter. The rule is simple. UI takes user action and sends it to the controller. And at the end it shows result taken from controller to the user. UI can be implemented according to MVC, MVP, MVVM or another approach. Control: Handles business logic of the application. Takes info from user and sends it to DB layer (DAO or ORM framework) and vice versa. Abstraction type of controller may vary (a separate control and business layers for example) according to the application parameters or development patterns (MVC, MVP, MVVM, ...) but the main idea remains the same. Data Access: Database handling layer of the application. It may contain entity definitions, ORM framework or DB connection codes having SQL sentences, according to the abstraction decision. Its role is getting data from controller, performing data operation on database and sending results again to controller (if result exists). Database independence is a very important plus for this layer, which brings flexibility. Vertical layers: Security: Security is a general concept, which includes user authorization, user authentication, user role management, network or SQL injection attack prevention, data recovery etc but we grouped that items generally as "Security". Those issues must be handled for each horizontal layer if you want to have a fully secure software. Logging: Logging is very important for maintenance and reporting in enterprise applications. An easily configurable, parameterized, readable, flexible and correctly implemented logging layer for all horizontal layers is very useful for both developers and users. i18n: i18n (internationalization) brings multiple language support and user interface text changing capability without rebuilding code. It may also have localization property which can handle number, currency and date formatting. So i18n provides flexibility on appliations and should not be ignored. Exception Handling: Exception handling may also be included in "Security" or "Logging" layers. We should not use only general exception classes or absorb all exceptions if we want to develop a quality software. We sometimes need our special exception classes with their specific behaviours for better security, logging and application performance. If you design an architecture which has that layer structure considering software quality factors (which are told here), you will probably have a successful enterprise software application.

When is it appropriate to use aspect oriented architecture? I think the only honest answer to this question is that it depends on the context for which the question is being asked. There really are no hard and fast rules regarding the selection of an architectural model(s) for a project because each model provides good and bad benefits. Every system is built with a unique requirements and constraints. This context will dictate when to use one type of architecture over another or in conjunction with others. To me aspect oriented architecture models should be a sub-phase in the architectural modeling and design process especially when creating enterprise level models. Personally, I like to use this approach to create a base architectural model that is defined by non-functional requirements and system quality attributes. This general model can then be used as a starting point for additional models because it is targets all of the business key quality attributes required by the system. Aspect oriented architecture is a method for modeling non-functional requirements and quality attributes of a system known as aspects. These models do not deal directly with specific functionality. They do categorize functionality of the system. This approach allows a system to be created with a strong emphasis on separating system concerns into individual components. These cross cutting components enables a systems to create with compartmentalization in regards to non-functional requirements or quality attributes. This allows for the reduction in code because an each component maintains an aspect of a system that can be called by other aspects. This approach also allows for a much cleaner and smaller code base during the implementation and support of a system. Additionally, enabling developers to develop systems based on aspect-oriented design projects will be completed faster and will be more reliable because existing components can be shared across a system; thus, the time needed to create and test the functionality is reduced. Example of an effective use of Aspect Oriented Architecture In my experiences, aspect oriented architecture can be very effective with large or more complex systems. Typically, these types of systems have a large number of concerns so the act of defining them is very beneficial for reducing the system’s complexity because components can be developed to address each concern while exposing functionality to the other system components. The benefits to using the aspect oriented approach as the starting point for a system is that it promotes communication between IT and the business due to the fact that the aspect oriented models are quality attributes focused so not much technical understanding is needed to understand the model. An example of this can be in developing a new intranet website. Common Intranet Concerns: Error Handling Security Logging Notifications Database connectivity Example of a not as effective use of Aspect Oriented Architecture Again in my experiences, aspect oriented architecture is not as effective with small or less complex systems in comparison. There is no need to model concerns for a system that has a limited amount of them because the added overhead would not be justified for the actual benefits of creating the aspect oriented architecture model. Furthermore, these types of projects typically have a reduced time schedule and a limited budget. The creation of the Aspect oriented models would increase the overhead of a project and thus increase the time needed to implement the system. An example of this is seen by creating a small application to poll a network share for new files and then FTP them to a new location. The two primary concerns for this project is to monitor a network drive and FTP files to a new location. There is no need to create an aspect model for this system because there will never be a need to share functionality amongst either of these concerns. To add to my point, this system is so small that it could be created with just a few classes so the added layer of componentizing the concerns would be complete overkill for this situation. References: Brichau, Johan; D'Hondt, Theo. (2006) Aspect-Oriented Software Development (AOSD) - An Introduction. Retreived from: http://www.info.ucl.ac.be/~jbrichau/courses/introductionToAOSD.pdf

For people in hurry get the code from Github. In continuation of my earlier blog on spring-test-mvc junit testing Spring Security layer with InMemoryDaoImpl, in this blog I will discuss how to use achieve method level access control. Please follow the steps in this blog to setup spring-test-mvc and run the below test case. mvn test -Dtest=com.example.springsecurity.web.controllers.SecurityControllerTest The JUnit test case looks as below, @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(loader = WebContextLoader.class, value = { "classpath:/META-INF/spring/services.xml", "classpath:/META-INF/spring/security.xml", "classpath:/META-INF/spring/mvc-config.xml" }) public class SecurityControllerTest { @Autowired CalendarService calendarService; @Test public void testMyEvents() throws Exception { Authentication auth = new UsernamePasswordAuthenticationToken("[email protected]", "user1"); SecurityContext securityContext = SecurityContextHolder.getContext(); securityContext.setAuthentication(auth); calendarService.findForUser(0); SecurityContextHolder.clearContext(); } @Test(expected = AuthenticationCredentialsNotFoundException.class) public void testForbiddenEvents() throws Exception { calendarService.findForUser(0); } } @Test(expected=AccessDeniedException.class) public void testWrongUserEvents() throws Exception { Authentication auth = new UsernamePasswordAuthenticationToken("[email protected]", "user2"); SecurityContext securityContext = SecurityContextHolder.getContext(); securityContext.setAuthentication(auth); calendarService.findForUser(0); SecurityContextHolder.clearContext(); } If you notice, if the user did not login or if the user is trying to access another users information it will throw an exception. The interface access control is as below, public interface CalendarService { @PreAuthorize("hasRole('ROLE_ADMIN') or principal.id == #userId") List findForUser(int userId); } The PreAuthorize only works on interface so that any implementation that implements this interface has this access control. I hope this blog helps you.

The Lamborghini of Key-Value stores Camel is the best of bread Integration framework and in this post I'm going to show you how to make it even more powerful by leveraging another great project - Redis. Camel 2.11 is on its way to be released soon with lots of new features, bug fixes and components. Couple of these new components are authored by me, redis-component being my favourite one. Redis - a ligth key/value store is an amazing piece of Italian software designed for speed (same as Lamborghini - a two-seater Italian car designed for speed). Written in C and having an in-memory closer to the metal nature, Redis performs extremely well (Lamborgini's motto is "Closer to the Road"). Redis is often referred to as a data structure server since keys can contain strings, hashes, lists and sorted sets. A fast and light data structure server is like a super sportscars for software engineers - it just flies. If you want to find out more about Redis' and Lamborghini's unique performance characteristics google around and you will see for yourself. Getting started with Redis is easy: download, make, and start a redis-server. After these steps, you ready to use it from your Camel application. The component uses internally Spring Data which in turn uses Jedis driver, but with possibility to switch to other Redis drivers. Here are few use cases where the camel-redis component is a good fit: Idempotent Repository The term idempotent is used in mathematics to describe a function that produces the same result if it is applied to itself. In Messaging this concepts translates into the a message that has the same effect whether it is received once or multiple times. In Camel this pattern is implemented using the IdempotentConsumer class which uses an Expression to calculate a unique message ID string for a given message exchange; this ID can then be looked up in the IdempotentRepository to see if it has been seen before; if it has the message is consumed; if its not then the message is processed and the ID is added to the repository. RedisIdempotentRepository is using a set structure to store and check for existing Ids. ${in.body.id} Caching One of the main uses of Redis is as LRU cache. It can store data inmemory as Memcached or can be tuned to be durable flushing data to a log file that can be replayed if the node restarts.The various policies when maxmemory is reached allows creating caches for specific needs: volatile-lru remove a key among the ones with an expire set, trying to remove keys not recently used. volatile-ttl remove a key among the ones with an expire set, trying to remove keys with short remaining time to live. volatile-random remove a random key among the ones with an expire set. allkeys-lru like volatile-lru, but will remove every kind of key, both normal keys or keys with an expire set. allkeys-random like volatile-random, but will remove every kind of keys, both normal keys and keys with an expire set. Once your Redis server is configured with the right policies and running, the operation you need to do are SET and GET: SET keyOne valueOne Interap pub/sub with Redis Camel has various components for interacting between routes: direct: provides direct, synchronous invocation in the same camel context. seda: asynchronous behavior, where messages are exchanged on a BlockingQueue, again in the same camel context. vm: asynchronous behavior like seda, but also supports communication across CamelContext as long as they are in the same JVM. Complex applications usually consist of more than one standalone Camel instances running on separate machines. For this kind of scenarios, Camel provides jms, activemq, combination of AWS SNS with SQS, for messaging between instances. Redis has a simpler solution for the Publish/Subscribe messaging paradigm. Subscribers subscribes to one or more channels, by specifying the channel names or using pattern matching for receiving messages from multiple channels. Then the publisher publishes the messages to a channel, and Redis makes sure it reaches all the matching subscribers. PUBLISH testChannel Test Message Other usages Guaranteed Delivery: Camel supports this EIP using JMS, File, JPA and few other components. Here Redis can be used as lightweight key-value persistent store with its transaction support. The Claim Check from the EIP patterns allows you to replace message content with a claim check (a unique key), which can be used to retrieve the message content at a later time. The message content can be stored temporarily in Redis. Redis is also very popular for implementing counters, leaderboards, tagging systems and many more functionalities. Now, with two swiss army knives under your belt, the integrations to make are limited only by your imagination.