Asynchronous Naiveté

This post has been heavily edited. Initially it was describing some behavior I didn't expect using core.async in both ClojureScript and Clojure. It turns out the behavior I observed in ClojureScript was a bug, that has now been fixed. The Clojure and ClojureScript versions can still behave differently than each other, depending on the lifetime of the main thread in Clojure. I have rewritten this to focus on that.

I created a buffered channel in ClojureScript, and tried to put 3 values into it:

(let [c (chan 1)]
  (go
   (.log js/console (<! c)))
  (go
   (>! c 1)
   (.log js/console "put one")
   (>! c 2)
   (.log js/console "put two")
   (>! c 3)
   (.log js/console "Put all my values on the channel")))

The result:

put one
put two
1

Running the equivalent code in a Clojure REPL

  (let [c (chan 1)]
    (go
     (println (<! c)))
    (go
     (>! c 1)
     (println "put one")
     (>! c 2)
     (println "put two")
     (>! c 3)
     (println "put all of my values on the channel")))

Gives the same 3 lines printed out though which line has the number 1 can change from invocation to invocation.

put one
1
put two  

When I put this in a -main method and ran it with lein run, I got no results at all.

Adding Thread/sleep the -main function causes the expected results to return. Though, again which order the 1 gets printed in can vary.

(defn -main [& args]
  (let [c (chan 1)]
    (go
     (println (<! c)))
    (go
     (>! c 1)
     (println "put one")
     (>! c 2)
     (println "put two")
     (>! c 3)
     (println "put all of my values on the channel"))
    (Thread/sleep 1000)))

put one
put two  
1

Putting the main thread to sleep gives the go blocks time to execute. For confirmation of what is happening, we can check what thread each operation is happening on.

(defn -main [& args]
  (let [c (chan 1)]
    (go
     (println (<! c))
     (println  (str "out " (.getId (Thread/currentThread)))))
    (go
     (>! c 1)
     (println "put one")
     (println (str "in1 " (.getId (Thread/currentThread))))
     (>! c 2)
     (println "put two")
     (println (str "in2 " (.getId (Thread/currentThread))))
     (>! c 3)
     (println "put all of my values on the channel"))
    (Thread/sleep 1000)
    (println (str "main  " (.getId (Thread/currentThread))))))

When I ran it I had the first go block on thread 10, the second on 11 and the main thread of execution on thread 1.

put one
in1 11
put two
in2 11
1
out 10
main  1

If we remove the Thread/sleep expression, but leave the calls to println, our output is:

main  1

This is just demo code. I don't know when in production systems your main thread would be likely to exit before asynchronous tasks are complete.

I do think it is interesting that the blocks of code I have run have executed in the same order every time in ClojureScript, but that the order can vary when running on the JVM. I do not know if that would be true of more complicated examples or on different browsers. I have done all of my tests with Google Chrome.

Simulated stock ticker with core.async

I created a simulated stock ticker in Clojure using core.async.

(ns ticker.core
  (:require [clojure.core.async
             :refer [chan ! timeout go]
              :as async]))

In core.async functions in your program communicate over channels. There are several ways to put items into a channel and take them off, but in this project only uses >! and <!. These functions must be called within a 'go block'.

<! takes a value off of a channel when one is available. If no value is available, execution within the go block is suspended until a value becomes available, at which time execution resumes.

>! does just the opposite. It puts a value onto a channel when a channel can accept a value, and suspends if the channel is not ready to accept a value. An unbuffered channel can only accept a new value when another block is trying to take a value off of the channel. Channels can use buffers to accept values before readers are ready for them, but this project doesn't use them.

The other core.async function that I refer is timeout. timeout creates a channel that closes after a specified number of milliseconds. Reading from a closed channel immediately returns nil. As the name suggests, this can be used to implement timeouts, but it can also be used to create delays.

I have a couple of utility functions for adding some randomness to the price and timing of transactions. I also have a function that creates a map to represent each transaction.

(defn adjust-price [old-price]
  (let  [numerator (- (rand-int 30) 15)
         adjustment (* numerator 0.01M)]
    (+ old-price adjustment)))

(defn random-time [t]
  (* t (+ 1 (rand-int 5))))

(defn new-transaction [symbol price]
  {:symbol symbol
   :time (java.util.Date.) 
   :price price})

The make-ticker function takes a stock symbol, a minimum number of milliseconds between transactions and a starting price. It returns a channel that will have a new transactions placed on it after a random interval as long as there is a listener to take the transactions off of the channel.

(defn make-ticker [symbol t start-price]
  (let [c (chan)]
    (go
     (loop [price start-price]
       (let [new-price (adjust-price price)]
         (<! (timeout (random-time t)))
         (>! c (new-transaction symbol new-price))
         (recur new-price))))
    c))

This function creates a channel, and then sets up an infinite loop that puts messages on to that channel. In terms of execution order what really happens is the channel is created and returned, a new price is calculated and the timeout is encountered. After the timeout has expired, a new value is put on the channel and we repeat the loop.

One thing that can get tricky with core.async is the lifetime of the channels. It is important to create the main channel 'c' outside of the loop because it needs to exist for the entire lifetime of the ticker. If it is created inside of the loop, each message will be on a new channel.

The timeout channel must be created inside of the loop. Each timeout is for a single use. If the timeout were created outside of the go loop, we would wait for the timeout channel to close during the first iteration. Later attempts to read the same channel would return immediately.

I created a collection of stocks symbols along with arbitrary values to use for the time interval and starting price.

(def stocks [ ;; symbol min-interval starting-price
             ["AAPL" 1400 537 ]
             ["AMZN" 4200 345]
             ["CSCO" 400  22]
             ["EBAY" 1200 55]
             ["GOOG" 8200 1127]
             ["IBM" 2200  192]
             ["MSFT" 500 40]
             ["ORCL" 1000 39]
             ["RHT" 10200  53]
             ["T" 600 35]])

Each stock symbol will have its transactions created on its own channel. For the ticker, we want to create a single channel that combines the outputs of each stock's channel. the merge function does exactly that, it takes several channels as inputs, and combines their outputs into a single channel.

(defn run-sim []
  (let [ticker (async/merge
                (map #(apply make-ticker %) stocks))]
    (go
     (loop [x 0]
       (when (< x 1000)
         (do (println (str x "-" (<! ticker)))
             (recur (inc x))))))))

This function creates channels for each of the stock symbols and combines their outputs into a channel called ticker. It then creates a loop within a go block that will run until 1000 transactions have been printed out.

I chose to call the merge function with async/merge instead of bringing merge in with :refer. The other core.async functions have names that make it clear that they pertain to core.async. Merge is a name that could apply to lots of different things, so I wanted to be explicit.

Ohms Law Calculator in Om

Reading David Nolen's tweets and blog posts about his new ClojureScript UI library, Om got me excited to try it out. If you haven't read about it yet, The Future of JavaScript MVC Frameworks is a good place to start.

David has also written a couple of tutorials. I was working through the introductory tutorial. I stopped short of the section on "Higher Order Components" to see if I could I apply what I had learned to that point.

I created a very simple Ohm's Law calculator. I borrowed liberally from David's code but there are enough differences that a comparison may be informative. I have put the code out on github.

I hope this is a useful supplement to David's tutorial, but it may make little sense on its own.

With Om you build your user interface in components. Each component is defined with om/root. root takes a component constructor function, an associative data structure that contains application data, and a map of options for the rendering functions, one of which must be the target dom element to be rendered.

(def app-state (atom {:result ""}))

(om/root
calculator-view
  app-state
  {:target (. js/document (getElementById "app"))})

The function passed to om/root takes as arguments application state and what the documentation calls the "owning pure node" and returns a reified object that must implement either IRender or IRenderState, and may implement other life cycle protocols. You will see second parameter is called 'owner' which is created by Om. It is this owner that contains component local state, as you will find in the entry-pane function later on.

Components built with om/root may be composed of smaller components. These subcomponents are created with the om/build function. Like root, build takes a two argument function that returns an object that implements either IRender or IRenderState and optionally other protocols. Build also needs to be passed in the state the component requires, and may be passed a map of options.

In David's contact list example, the display for each contact was rendered by its own subcomponent. The state for each was a map containing details for a single contact. He constructed these components with om/build-all, passing it his constructor function and a vector of contacts. om/build-all maps om/build over the vector of individual contacts.

In my example, I am constructing two subcomponents, neither of which display a sequence of data, so I call om/build twice.

(defn calculator-view [app owner]
  (reify
    om/IRender
    (render  [this]
      (dom/div nil
               (om/build entry-pane app)
               (om/build result-pane app)))))

(defn result-pane [app owner]
  (reify
    om/IRender
    (render [this]
      (dom/div #js {:className "result-pane"}
               (dom/label #js {:className "result"} (:result app))))))

In David's example, his components both implemented IRenderState. That was because his contact-view created a core.async channel that was placed in each component's state to allow for delete messages to be sent from the child components to the parent. My calculator-view and result-pane share nothing except the app-state which is available in IRender.

My entry-pane function does rely on component state. I copied David's implementation of handling button clicks with a channel and the keyboard events with an anonymous function that calls a handle-change function. The channel and the values of the text box are all state that is needed only within the component.

(defn entry-pane [app owner]
  (reify
    om/IInitState
    (init-state [_]
      {:calculate (chan)
       :ohms ""
       :amps ""
       :volts ""})
    om/IWillMount
    (will-mount [_]
      (let [calculate (om/get-state owner :calculate)]
        (go (loop []
              (let [inputs (<! calculate)]
                (om/transact! app
                              (fn [xs]
                                (assoc xs :result (do-calc inputs))))
                (recur))))))
    om/IRenderState
    (render-state [this state]
      (dom/div #js {:className "entry-pane"}
       (dom/h3 nil "Ohm's Law Calculator")
       (dom/label nil "Resistance")
       (dom/input #js {:type "text" :ref "ohms"
                       :value (:ohms state)
                       :onChange #(handle-change % owner :ohms state)})
       (dom/label nil "Current")
       (dom/input #js {:type "text" :ref "amps"
                       :value (:amps state)
                       :onChange #(handle-change % owner :amps state)})
       (dom/label nil "Voltage")
       (dom/input #js {:type "text" :ref "volts"
                       :value (:volts state)
                       :onChange #(handle-change % owner :volts state)})
       (dom/button #js {:className "button"
                        :onClick
                        (fn [e] (put! (:calculate state) state))}
                   "Calculate")))))

When the user clicks the calculate button the click handler puts the state map onto the calculate channel. The go block in the IWillMount implementation takes the state map from the channel and passes it to the calculate function which returns a string. The :result is set in the app state using the om/transact!. This mutates the state and triggers a re-render.

The handle-change function calls om/set-state! to update the local state with the data entered by the user. set-state! mutates the data and triggers a re-render just like transact! does. The difference is set-state! operates on state contained within a component.

(defn handle-change [e owner key state]
  (let [value (.. e -target -value)
        text (key state)
        allowed (set (seq (str (range 10))))]
    (if (every? allowed (seq value))
      (om/set-state! owner key value)
      (om/set-state! owner key text))))

The concepts in Om are still new to me. I have explained my rationale for the code I have written. If you are aware of any mistakes, please leave a comment letting me know where I have gone wrong.

ML Class Notes: Lesson 1 - Introduction

I am taking the Machine Learning class at Coursera. These are my notes on the material presented by Professor Ng.

The first lesson introduces a number of concepts in machine learning. There is no code to show until the first algorithm is introduced in the next lesson.

Machine learning grew out of AI research. It is a field of study that gives computers the ability to learn algorithms and processes that can not be explicitly programmed. Computers could be programmed to do simple things, but doing more complicated things required the computer learn itself. A well posed learning program is said to learn some task if its performance improves with experience.

Machine Learning is used for a lot of things including data mining in business, biology and engineering; performing tasks that can't be programmed by hand like piloting helicopters or computer vision; self-customizing programs like product recommendations; and as a model to try to understand human learning.

Two of the more common categories of machine learning algorithms are supervised and unsupervised learning. Other categories include reinforcement learning and recommender systems, but they were not described in this lesson.

Supervised Learning

In supervised learning the computer taught to make predictions using a set of examples where the historical result is already known. One type of supervised learning tasks is regression where the predicted value is in a continuous range (the example given was predicting home prices). Other supervised learning algorithms perform classification where examples are sorted into two or more buckets (the examples given were of email, which can be spam or not spam; and tumor diagnosis which could be malignant or benign.)

Unsupervised Learning

In unsupervised learning, the computer must teach itself to perform a task because the "correct" answer is not known. A common supervised learning task is clustering. Clustering is used to group data points into different categories based on their similarity to each other. Professor Ng gave the the example of Google News, which groups related news articles, allowing you to select accounts of the same event from different news sources.

The unsupervised learning discussion ended with a demonstration of an algorithm that had been used to solve the "cocktail party problem", where two people were speaking at the same time in the same room, and were recorded by two microphones in different parts of the room. The clustering algorithm was used to determine which sound signals were from each speaker. In the initial recordings, both speakers could be heard on both microphones. In the sound files produced by the learning algorithm, each output has the sound from one speaker, with the other speaker almost entirely absent.

Take the Machine Learning Class at Coursera

Coursera is offering its Machine Learning course again, beginning March 8, and I highly recommend it. You already know the obvious, that it is a course on an incredibly timely career skill and it is free, but until you take the course you can't know just how good the course really is.

You will learn how to write algorithms to perform linear regression, logistic regression, neural networks, clustering and dimensionality reduction. Throughout the course Professor Ng explains the techniques that are used to prepare data for analysis, why particular techniques are used, and how to determine which techniques are most useful for a particular problem.

In addition to the explanation of what and why, there is an equal amount of explaining how. The 'how' is math, specifically linear algebra. From the first week to the last, Ng clearly explains the mathematical techniques and equations that apply to each problem, how the equations are represented with linear algebra, and how to implement each calculation in Octave or Matlab.

The course has homework. Each week, there is a zip file that contains a number of incomplete matlab files that provide the structure for the problem to be solved, and you need to implement the techniques from the week's lessons. Each assignment includes a submission script that is run from the command line. You submit your solution, and it either congratulates you for getting the right answer, or informs you if your solution was incorrect.

It is possible to view all of the lectures without signing up for the class. Don't do that. Sign up for the class. Actually signing up for the class gives you a schedule to keep to. It also allows you to get your homework checked. When you watch the lectures, you will think you understand the material; until you have done the homework you really don't. As good as the teaching is, the material is still rigorous enough that it will be hard to complete if you are not trying to keep to a schedule. Also, if you complete the course successfully, you will be able to put it on your resume and LinkedIn profile.

You have the time. When I took the class, there was extra time built in to the schedule to allow people who started the course late to stay on pace. Even if you fall behind, the penalty for late submission is low enough that it is possible to complete every assignment late and still get a passing grade in the course.

I am going to take the course again. I want to make review the material. I also want to try to implement the homework solutions in Clojure, in addition to Octave. I will be posting regularly about my progress.

You may also be able to find a study group in your area. I decided to retake the course when I found out that there was going to be a meetup group in my area. Even without a local group, the discussion forums are a great source of help throughout the class. The teaching assistants and your classmates provide a lot of guidance when you need it.

Refactoring Blob Store Access

My last post was a translation of Microsoft's examples of accessing the Windows Azure blob storage from Java to Clojure. The post consisted of a series of interop calls without any context.

I thought it would be interesting to see how they looked in an application. The project I am building is a file backup program.

In the Microsoft examples, each example began with creating a connection string and a reference to the blob store container. In my translation, I just created set up the reference to the container once at the top of the file.

blobstore.clj

I decided to hold the reference to the container in a closure. The container function takes a connection string and container name, uses the Azure sdk classes to build the reference to the container, create the container in Azure if it doesn't already exists, and returns a map of functions that can be executed on the container.

(defn container [conn-str container-name]
  (let [ctr
        (-> conn-str
            CloudStorageAccount/parse
            .createCloudBlobClient
            (.getContainerReference container-name))]
    (.createIfNotExist ctr)

    {:upload (fn [{:keys [file target]}]
               (let [blob-ref (.getBlockBlobReference ctr target)]
                 (with-open [r (FileInputStream. file)]
                   (.upload blob-ref r (.length file)))))

     :download (fn [{:keys [blob target]}]
                 (do
                   (.mkdirs (.getParentFile (File. target)))
                   (with-open [w (FileOutputStream. target)]
                   (.download blob w))))

     :find-blob (fn [blobname]
                  (.getBlockBlobReference ctr blobname))

     :delete (fn [blob]
               (.delete blob))

     :remove-container (fn []
                         (.delete ctr))

     :blob-seq (fn []
                 (filter #(instance? CloudBlockBlob %)
                  (tree-seq
                   (fn [f] (not (instance? CloudBlockBlob f)))
                   (fn [f] (.listBlobs f))
                   ctr)))

     :delete-container (fn []
                         (.delete ctr))
     }))

There is one other change I want to call attention to. Testing the code in the repl, I discovered that the FileOutputStream in the download function was keeping a connection to the file on the file system. I assume the FileInputStream in the upload function works the same way. To fix this, I used the with-open macro to cleanup the streams when I was done using them.

I created a new code file to hold this function. I wanted my core.clj file to make the decisions about what needed to be done, but to know nothing about how anything would be done.

core.clj

My ns declaration looks like:

(ns filer.core
  (:gen-class)
  (:require [filer.config :as config]
            [filer.blobstore :as store]
            [filer.filestore :as files]))

Jumping to the bottom of the file the main function causes one of three actions to be taken: The default is for file system folders specified in a configuration file to be backed up to the blob store; a specified blob container can be downloaded to a restore folder specified in the config file; or a blob container can be deleted.

(defn -main [& args]
  (cond
   (= "delete" (first args))
     (delete-blobs (store/container config/conn-str (second args)))
   (= "restore" (first args))
    (restore-folder config/restore-folder (store/container config/conn-str (second args)))
   :default
     (doseq [p config/back-folders]
       (backup-folder p (make-container p)))))

The backup function operates on a collection of folders to backup. Each root folder is stored in a separate container in the blob store. To create a naming system for my backup containers, I added a function to my config.clj file that returns a container name to use based on the file system folder and the date.

The call to subs in the upload-settings function is to strip off the part of the file name that pertains to the root folder, which is already represented by the name of the container the file is being put into. Looking at it now, this definitely violates my goal of separating what to do from how to do it. I may want to move my whole upload settings function into the blobstore.clj file, but certainly the details of translating file system names to blob store names does not belong here.

(defn make-container [root-folder]
  (store/container config/conn-str
    (config/container-name root-folder)))

(defn upload-settings [f root-folder]
  {:file f
   :target (subs (.toString f) (inc (count root-folder)))})

(defn upload-file [file container root-folder]
  ((:upload container) (upload-settings file root-folder)))

(defn backup-folder [folder container]
  (doseq [f (files/all-files folder)]
    (upload-file f container folder)))

The restore function is similar to the backup function, in that both walk through a sequence of files on a source system, determine their name on the destination system and then call the appropriate function on the container.

The primary difference is that backing up files is done for a collection of root folders, which each get their own container, so I need a function to execute for each folder. The program is set up to only restore a single container specified as a command line argument. The -main function gets the single reference to the container, and passes it to the restore folder.

(defn get-destination [blob folder]
  (files/ms-name
   (str folder "/" (.getName blob))))

(defn download-settings [blob folder]
  {:blob blob
   :target (get-destination blob folder)})

(defn restore-folder [folder container]
  (doseq [f ((:blob-seq container))]
    ((:download container) (download-settings f folder))))

The delete function is the simplest of all. Deleting a container also deletes all of the files in it. The delete container function could be called straight from -main, but for now it is its own function.

(defn delete-blobs [ctr]
  ((:delete-container ctr)))

Thoughts about this design

Creating the container in one place and then returning a map of functions that reference the container works pretty well. The one bit of awkwardness is that it means that all of the functions have to be invoked with double parentheses. The inner set is for the lookup on the map, the outer set invokes the returned function.

I can make the code look better by binding the function to a symbol in a let, and then using that symbol in the function call. For the restore-folder function it should also help performance some.

(defn restore-folder [folder container]
  (let [downfn (:download container)]
    (doseq [f ((:blob-seq container))]
      (downfn (download-settings f)))))

Now I look up the function only once, and then use the same function for every file I download. The cost of looking up a function compared to the cost of downloading a file is minimal, so I will think about it for a while, and keep the version I decide looks better.

Using a .clj file for my configuration file was a pretty obvious choice. Clojure is a superset of edn, so I could probably make use of tagged elements, but I just used def statements. The functions to provide standardized folder names seemed right at home here.

All of the calls in this program are synchronous calls. In many applications it makes sense to make calls out to the file system or the cloud asynchronous. For this application, however, I don't think it would add much. This is a program that is meant to be called from the command line, with no user interface to block. At one point, I did have an asynchronous version of my upload function but I didn't think it added much besides complexity.

Conclusion

Writing this post I found several errors in my code, and a couple of ways that I could have expressed things differently. Adding let bindings for the function lookups seems obvious now, but I hadn't yet thought of it an hour ago.

Thank you to anyone who reads this post. I hope you have gotten some benefit from seeing my thought process. I will be doing more posts like this in the future. And if you don't benefit from these posts, sorry about the noise. :)

Azure Blob Storage from Clojure

I need to write some files to Windows Azure Blob Storage. Windows Azure provides libraries in several languages and there is also a REST API.

Each version of the library comes with very good documentation. It explains how blob storage works, explains how to create an account and walks you through adding, downloading and deleting the blobs. The Java version is at http://www.windowsazure.com/en-us/documentation/articles/storage-java-how-to-use-blob-storage/

The project I am working on is in Clojure. Translating the examples is pretty straight forward.

The Java libraries are on Maven Central, so I added a dependency to my project.clj.

[com.microsoft.windowsazure/microsoft-windowsazure-api "0.4.6"]

The classes we need to import classes from the windowsazure services.core.storage package and the services.blob.client packages. The examples also use several classes from java.io. I also created a config.clj file in my src directory to hold my account name and key. So, my core.clj ns macro is:

(ns azure-blob-test.core
  (:require [azure-blob-test.config :as config])
  (:import [com.microsoft.windowsazure.services.core.storage CloudStorageAccount]
           [com.microsoft.windowsazure.services.blob.client CloudBlobClient CloudBlobContainer CloudBlockBlob CloudBlob BlobContainerPermissions BlobContainerPublicAccessType]
           [java.io File FileInputStream FileOutputStream]))

Each of the examples begin with creating a reference to a blob container. I think they are doing that so that is someone just looks at how to add a blob, they have a complete example. Really, you just create it once:

(def conn-str
  (str "DefaultEndpointsProtocol=http;"
         "AccountName=" config/account-name
         ";AccountKey=" config/account-key ";"))

(def container
  (-> conn-str
      CloudStorageAccount/parse
      .createCloudBlobClient
      (.getContainerReference "mycontainer")))

(.createIfNotExist container)

Setting the container permissions looks clunky to me in both languages. Setting public access means that people do not need an account key to access the folder. Actually, not what I want for my application, but for the sake of completeness:

(let [container-permissions (BlobContainerPermissions.)]
  (.setPublicAccess container-permissions BlobContainerPublicAccessType/CONTAINER)
  (.uploadPermissions container container-permissions))

For testing the upload, I decided to use project.clj, since I know it will be in the root folder when I am running in the REPL.

(let [blob-ref (.getBlockBlobReference container "project.clj")
      source-file (File. "project.clj")]
  (.upload blob-ref (FileInputStream. source-file) (.length source-file)))

Rather than print out a list of blobs in a container, I decided to write a function that returns a sequence of blobs. The filter checks for instances of CloudBlob to exclude virtual directories. And then I use that function to download all of the blobs. I append .bak to the filenames because I don't want to overwrite my project.clj, even if it is the same (superstitious?).

(defn blob-list [container]
  (filter #(instance? CloudBlob %) (.listBlobs container)))

(doseq [blob (blob-list container)]
  (.download blob (FileOutputStream. (str (.getName blob) ".bak"))))

And finally to clean up:

;; Delete the blob
(.delete (.getBlockBlobReference container "project.clj"))

;; Delete container
(.delete container)